chiark / gitweb /
bus-message: remove shadow warning with log_debug_bus_message()
[elogind.git] / src / libelogind / sd-bus / sd-bus.c
1 /*-*- Mode: C; c-basic-offset: 8; indent-tabs-mode: nil -*-*/
2
3 /***
4   This file is part of systemd.
5
6   Copyright 2013 Lennart Poettering
7
8   systemd is free software; you can redistribute it and/or modify it
9   under the terms of the GNU Lesser General Public License as published by
10   the Free Software Foundation; either version 2.1 of the License, or
11   (at your option) any later version.
12
13   systemd is distributed in the hope that it will be useful, but
14   WITHOUT ANY WARRANTY; without even the implied warranty of
15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16   Lesser General Public License for more details.
17
18   You should have received a copy of the GNU Lesser General Public License
19   along with systemd; If not, see <http://www.gnu.org/licenses/>.
20 ***/
21
22 #include <endian.h>
23 #include <stdlib.h>
24 #include <unistd.h>
25 #include <netdb.h>
26 #include <poll.h>
27 #include <sys/mman.h>
28 #include <pthread.h>
29
30 #include "util.h"
31 #include "macro.h"
32 #include "strv.h"
33 #include "missing.h"
34 #include "def.h"
35 #include "cgroup-util.h"
36 #include "bus-label.h"
37
38 #include "sd-bus.h"
39 #include "bus-internal.h"
40 #include "bus-message.h"
41 #include "bus-type.h"
42 #include "bus-socket.h"
43 #include "bus-kernel.h"
44 #include "bus-control.h"
45 #include "bus-objects.h"
46 #include "bus-util.h"
47 #include "bus-container.h"
48 #include "bus-protocol.h"
49 #include "bus-track.h"
50 #include "bus-slot.h"
51
52 #define log_debug_bus_message(m)                                         \
53         do {                                                             \
54                 sd_bus_message *_mm = (m);                               \
55                 log_debug("Got message type=%s sender=%s destination=%s object=%s interface=%s member=%s cookie=%" PRIu64 " reply_cookie=%" PRIu64 " error=%s", \
56                           bus_message_type_to_string(_mm->header->type), \
57                           strna(sd_bus_message_get_sender(_mm)),         \
58                           strna(sd_bus_message_get_destination(_mm)),    \
59                           strna(sd_bus_message_get_path(_mm)),           \
60                           strna(sd_bus_message_get_interface(_mm)),      \
61                           strna(sd_bus_message_get_member(_mm)),         \
62                           BUS_MESSAGE_COOKIE(_mm),                       \
63                           _mm->reply_cookie,                             \
64                           strna(_mm->error.message));                    \
65         } while (false)
66
67 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
68 static int attach_io_events(sd_bus *b);
69 static void detach_io_events(sd_bus *b);
70
71 static void bus_close_fds(sd_bus *b) {
72         assert(b);
73
74         detach_io_events(b);
75
76         if (b->input_fd >= 0)
77                 safe_close(b->input_fd);
78
79         if (b->output_fd >= 0 && b->output_fd != b->input_fd)
80                 safe_close(b->output_fd);
81
82         b->input_fd = b->output_fd = -1;
83 }
84
85 static void bus_reset_queues(sd_bus *b) {
86         assert(b);
87
88         while (b->rqueue_size > 0)
89                 sd_bus_message_unref(b->rqueue[--b->rqueue_size]);
90
91         free(b->rqueue);
92         b->rqueue = NULL;
93         b->rqueue_allocated = 0;
94
95         while (b->wqueue_size > 0)
96                 sd_bus_message_unref(b->wqueue[--b->wqueue_size]);
97
98         free(b->wqueue);
99         b->wqueue = NULL;
100         b->wqueue_allocated = 0;
101 }
102
103 static void bus_free(sd_bus *b) {
104         sd_bus_slot *s;
105
106         assert(b);
107         assert(!b->track_queue);
108
109         b->state = BUS_CLOSED;
110
111         sd_bus_detach_event(b);
112
113         while ((s = b->slots)) {
114                 /* At this point only floating slots can still be
115                  * around, because the non-floating ones keep a
116                  * reference to the bus, and we thus couldn't be
117                  * destructing right now... We forcibly disconnect the
118                  * slots here, so that they still can be referenced by
119                  * apps, but are dead. */
120
121                 assert(s->floating);
122                 bus_slot_disconnect(s);
123                 sd_bus_slot_unref(s);
124         }
125
126         if (b->default_bus_ptr)
127                 *b->default_bus_ptr = NULL;
128
129         bus_close_fds(b);
130
131         if (b->kdbus_buffer)
132                 munmap(b->kdbus_buffer, KDBUS_POOL_SIZE);
133
134         free(b->label);
135         free(b->rbuffer);
136         free(b->unique_name);
137         free(b->auth_buffer);
138         free(b->address);
139         free(b->kernel);
140         free(b->machine);
141         free(b->fake_label);
142         free(b->cgroup_root);
143         free(b->description);
144
145         free(b->exec_path);
146         strv_free(b->exec_argv);
147
148         close_many(b->fds, b->n_fds);
149         free(b->fds);
150
151         bus_reset_queues(b);
152
153         ordered_hashmap_free_free(b->reply_callbacks);
154         prioq_free(b->reply_callbacks_prioq);
155
156         assert(b->match_callbacks.type == BUS_MATCH_ROOT);
157         bus_match_free(&b->match_callbacks);
158
159         hashmap_free_free(b->vtable_methods);
160         hashmap_free_free(b->vtable_properties);
161
162         assert(hashmap_isempty(b->nodes));
163         hashmap_free(b->nodes);
164
165         bus_kernel_flush_memfd(b);
166
167         assert_se(pthread_mutex_destroy(&b->memfd_cache_mutex) == 0);
168
169         free(b);
170 }
171
172 _public_ int sd_bus_new(sd_bus **ret) {
173         sd_bus *r;
174
175         assert_return(ret, -EINVAL);
176
177         r = new0(sd_bus, 1);
178         if (!r)
179                 return -ENOMEM;
180
181         r->n_ref = REFCNT_INIT;
182         r->input_fd = r->output_fd = -1;
183         r->message_version = 1;
184         r->creds_mask |= SD_BUS_CREDS_WELL_KNOWN_NAMES|SD_BUS_CREDS_UNIQUE_NAME;
185         r->hello_flags |= KDBUS_HELLO_ACCEPT_FD;
186         r->attach_flags |= KDBUS_ATTACH_NAMES;
187         r->original_pid = getpid();
188
189         assert_se(pthread_mutex_init(&r->memfd_cache_mutex, NULL) == 0);
190
191         /* We guarantee that wqueue always has space for at least one
192          * entry */
193         if (!GREEDY_REALLOC(r->wqueue, r->wqueue_allocated, 1)) {
194                 free(r);
195                 return -ENOMEM;
196         }
197
198         *ret = r;
199         return 0;
200 }
201
202 _public_ int sd_bus_set_address(sd_bus *bus, const char *address) {
203         char *a;
204
205         assert_return(bus, -EINVAL);
206         assert_return(bus->state == BUS_UNSET, -EPERM);
207         assert_return(address, -EINVAL);
208         assert_return(!bus_pid_changed(bus), -ECHILD);
209
210         a = strdup(address);
211         if (!a)
212                 return -ENOMEM;
213
214         free(bus->address);
215         bus->address = a;
216
217         return 0;
218 }
219
220 _public_ int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
221         assert_return(bus, -EINVAL);
222         assert_return(bus->state == BUS_UNSET, -EPERM);
223         assert_return(input_fd >= 0, -EINVAL);
224         assert_return(output_fd >= 0, -EINVAL);
225         assert_return(!bus_pid_changed(bus), -ECHILD);
226
227         bus->input_fd = input_fd;
228         bus->output_fd = output_fd;
229         return 0;
230 }
231
232 _public_ int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
233         char *p, **a;
234
235         assert_return(bus, -EINVAL);
236         assert_return(bus->state == BUS_UNSET, -EPERM);
237         assert_return(path, -EINVAL);
238         assert_return(!strv_isempty(argv), -EINVAL);
239         assert_return(!bus_pid_changed(bus), -ECHILD);
240
241         p = strdup(path);
242         if (!p)
243                 return -ENOMEM;
244
245         a = strv_copy(argv);
246         if (!a) {
247                 free(p);
248                 return -ENOMEM;
249         }
250
251         free(bus->exec_path);
252         strv_free(bus->exec_argv);
253
254         bus->exec_path = p;
255         bus->exec_argv = a;
256
257         return 0;
258 }
259
260 _public_ int sd_bus_set_bus_client(sd_bus *bus, int b) {
261         assert_return(bus, -EINVAL);
262         assert_return(bus->state == BUS_UNSET, -EPERM);
263         assert_return(!bus_pid_changed(bus), -ECHILD);
264
265         bus->bus_client = !!b;
266         return 0;
267 }
268
269 _public_ int sd_bus_set_monitor(sd_bus *bus, int b) {
270         assert_return(bus, -EINVAL);
271         assert_return(bus->state == BUS_UNSET, -EPERM);
272         assert_return(!bus_pid_changed(bus), -ECHILD);
273
274         SET_FLAG(bus->hello_flags, KDBUS_HELLO_MONITOR, b);
275         return 0;
276 }
277
278 _public_ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
279         assert_return(bus, -EINVAL);
280         assert_return(bus->state == BUS_UNSET, -EPERM);
281         assert_return(!bus_pid_changed(bus), -ECHILD);
282
283         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ACCEPT_FD, b);
284         return 0;
285 }
286
287 _public_ int sd_bus_negotiate_timestamp(sd_bus *bus, int b) {
288         uint64_t new_flags;
289         assert_return(bus, -EINVAL);
290         assert_return(!IN_SET(bus->state, BUS_CLOSING, BUS_CLOSED), -EPERM);
291         assert_return(!bus_pid_changed(bus), -ECHILD);
292
293         new_flags = bus->attach_flags;
294         SET_FLAG(new_flags, KDBUS_ATTACH_TIMESTAMP, b);
295
296         if (bus->attach_flags == new_flags)
297                 return 0;
298
299         bus->attach_flags = new_flags;
300         if (bus->state != BUS_UNSET && bus->is_kernel)
301                 bus_kernel_realize_attach_flags(bus);
302
303         return 0;
304 }
305
306 _public_ int sd_bus_negotiate_creds(sd_bus *bus, int b, uint64_t mask) {
307         uint64_t new_flags;
308
309         assert_return(bus, -EINVAL);
310         assert_return(mask <= _SD_BUS_CREDS_ALL, -EINVAL);
311         assert_return(!IN_SET(bus->state, BUS_CLOSING, BUS_CLOSED), -EPERM);
312         assert_return(!bus_pid_changed(bus), -ECHILD);
313
314         if (b)
315                 bus->creds_mask |= mask;
316         else
317                 bus->creds_mask &= ~mask;
318
319         /* The well knowns we need unconditionally, so that matches can work */
320         bus->creds_mask |= SD_BUS_CREDS_WELL_KNOWN_NAMES|SD_BUS_CREDS_UNIQUE_NAME;
321
322         /* Make sure we don't lose the timestamp flag */
323         new_flags = (bus->attach_flags & KDBUS_ATTACH_TIMESTAMP) | attach_flags_to_kdbus(bus->creds_mask);
324         if (bus->attach_flags == new_flags)
325                 return 0;
326
327         bus->attach_flags = new_flags;
328         if (bus->state != BUS_UNSET && bus->is_kernel)
329                 bus_kernel_realize_attach_flags(bus);
330
331         return 0;
332 }
333
334 _public_ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
335         assert_return(bus, -EINVAL);
336         assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
337         assert_return(bus->state == BUS_UNSET, -EPERM);
338         assert_return(!bus_pid_changed(bus), -ECHILD);
339
340         bus->is_server = !!b;
341         bus->server_id = server_id;
342         return 0;
343 }
344
345 _public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
346         assert_return(bus, -EINVAL);
347         assert_return(bus->state == BUS_UNSET, -EPERM);
348         assert_return(!bus_pid_changed(bus), -ECHILD);
349
350         bus->anonymous_auth = !!b;
351         return 0;
352 }
353
354 _public_ int sd_bus_set_trusted(sd_bus *bus, int b) {
355         assert_return(bus, -EINVAL);
356         assert_return(bus->state == BUS_UNSET, -EPERM);
357         assert_return(!bus_pid_changed(bus), -ECHILD);
358
359         bus->trusted = !!b;
360         return 0;
361 }
362
363 _public_ int sd_bus_set_description(sd_bus *bus, const char *description) {
364         assert_return(bus, -EINVAL);
365         assert_return(bus->state == BUS_UNSET, -EPERM);
366         assert_return(!bus_pid_changed(bus), -ECHILD);
367
368         return free_and_strdup(&bus->description, description);
369 }
370
371 _public_ int sd_bus_set_allow_interactive_authorization(sd_bus *bus, int b) {
372         assert_return(bus, -EINVAL);
373         assert_return(!bus_pid_changed(bus), -ECHILD);
374
375         bus->allow_interactive_authorization = !!b;
376         return 0;
377 }
378
379 _public_ int sd_bus_get_allow_interactive_authorization(sd_bus *bus) {
380         assert_return(bus, -EINVAL);
381         assert_return(!bus_pid_changed(bus), -ECHILD);
382
383         return bus->allow_interactive_authorization;
384 }
385
386 static int hello_callback(sd_bus_message *reply, void *userdata, sd_bus_error *error) {
387         const char *s;
388         sd_bus *bus;
389         int r;
390
391         assert(reply);
392         bus = reply->bus;
393         assert(bus);
394         assert(bus->state == BUS_HELLO || bus->state == BUS_CLOSING);
395
396         r = sd_bus_message_get_errno(reply);
397         if (r > 0)
398                 return -r;
399
400         r = sd_bus_message_read(reply, "s", &s);
401         if (r < 0)
402                 return r;
403
404         if (!service_name_is_valid(s) || s[0] != ':')
405                 return -EBADMSG;
406
407         bus->unique_name = strdup(s);
408         if (!bus->unique_name)
409                 return -ENOMEM;
410
411         if (bus->state == BUS_HELLO)
412                 bus->state = BUS_RUNNING;
413
414         return 1;
415 }
416
417 static int bus_send_hello(sd_bus *bus) {
418         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
419         int r;
420
421         assert(bus);
422
423         if (!bus->bus_client || bus->is_kernel)
424                 return 0;
425
426         r = sd_bus_message_new_method_call(
427                         bus,
428                         &m,
429                         "org.freedesktop.DBus",
430                         "/org/freedesktop/DBus",
431                         "org.freedesktop.DBus",
432                         "Hello");
433         if (r < 0)
434                 return r;
435
436         return sd_bus_call_async(bus, NULL, m, hello_callback, NULL, 0);
437 }
438
439 int bus_start_running(sd_bus *bus) {
440         assert(bus);
441
442         if (bus->bus_client && !bus->is_kernel) {
443                 bus->state = BUS_HELLO;
444                 return 1;
445         }
446
447         bus->state = BUS_RUNNING;
448         return 1;
449 }
450
451 static int parse_address_key(const char **p, const char *key, char **value) {
452         size_t l, n = 0, allocated = 0;
453         const char *a;
454         char *r = NULL;
455
456         assert(p);
457         assert(*p);
458         assert(value);
459
460         if (key) {
461                 l = strlen(key);
462                 if (strncmp(*p, key, l) != 0)
463                         return 0;
464
465                 if ((*p)[l] != '=')
466                         return 0;
467
468                 if (*value)
469                         return -EINVAL;
470
471                 a = *p + l + 1;
472         } else
473                 a = *p;
474
475         while (*a != ';' && *a != ',' && *a != 0) {
476                 char c;
477
478                 if (*a == '%') {
479                         int x, y;
480
481                         x = unhexchar(a[1]);
482                         if (x < 0) {
483                                 free(r);
484                                 return x;
485                         }
486
487                         y = unhexchar(a[2]);
488                         if (y < 0) {
489                                 free(r);
490                                 return y;
491                         }
492
493                         c = (char) ((x << 4) | y);
494                         a += 3;
495                 } else {
496                         c = *a;
497                         a++;
498                 }
499
500                 if (!GREEDY_REALLOC(r, allocated, n + 2))
501                         return -ENOMEM;
502
503                 r[n++] = c;
504         }
505
506         if (!r) {
507                 r = strdup("");
508                 if (!r)
509                         return -ENOMEM;
510         } else
511                 r[n] = 0;
512
513         if (*a == ',')
514                 a++;
515
516         *p = a;
517
518         free(*value);
519         *value = r;
520
521         return 1;
522 }
523
524 static void skip_address_key(const char **p) {
525         assert(p);
526         assert(*p);
527
528         *p += strcspn(*p, ",");
529
530         if (**p == ',')
531                 (*p) ++;
532 }
533
534 static int parse_unix_address(sd_bus *b, const char **p, char **guid) {
535         _cleanup_free_ char *path = NULL, *abstract = NULL;
536         size_t l;
537         int r;
538
539         assert(b);
540         assert(p);
541         assert(*p);
542         assert(guid);
543
544         while (**p != 0 && **p != ';') {
545                 r = parse_address_key(p, "guid", guid);
546                 if (r < 0)
547                         return r;
548                 else if (r > 0)
549                         continue;
550
551                 r = parse_address_key(p, "path", &path);
552                 if (r < 0)
553                         return r;
554                 else if (r > 0)
555                         continue;
556
557                 r = parse_address_key(p, "abstract", &abstract);
558                 if (r < 0)
559                         return r;
560                 else if (r > 0)
561                         continue;
562
563                 skip_address_key(p);
564         }
565
566         if (!path && !abstract)
567                 return -EINVAL;
568
569         if (path && abstract)
570                 return -EINVAL;
571
572         if (path) {
573                 l = strlen(path);
574                 if (l > sizeof(b->sockaddr.un.sun_path))
575                         return -E2BIG;
576
577                 b->sockaddr.un.sun_family = AF_UNIX;
578                 strncpy(b->sockaddr.un.sun_path, path, sizeof(b->sockaddr.un.sun_path));
579                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l;
580         } else if (abstract) {
581                 l = strlen(abstract);
582                 if (l > sizeof(b->sockaddr.un.sun_path) - 1)
583                         return -E2BIG;
584
585                 b->sockaddr.un.sun_family = AF_UNIX;
586                 b->sockaddr.un.sun_path[0] = 0;
587                 strncpy(b->sockaddr.un.sun_path+1, abstract, sizeof(b->sockaddr.un.sun_path)-1);
588                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + 1 + l;
589         }
590
591         return 0;
592 }
593
594 static int parse_tcp_address(sd_bus *b, const char **p, char **guid) {
595         _cleanup_free_ char *host = NULL, *port = NULL, *family = NULL;
596         int r;
597         struct addrinfo *result, hints = {
598                 .ai_socktype = SOCK_STREAM,
599                 .ai_flags = AI_ADDRCONFIG,
600         };
601
602         assert(b);
603         assert(p);
604         assert(*p);
605         assert(guid);
606
607         while (**p != 0 && **p != ';') {
608                 r = parse_address_key(p, "guid", guid);
609                 if (r < 0)
610                         return r;
611                 else if (r > 0)
612                         continue;
613
614                 r = parse_address_key(p, "host", &host);
615                 if (r < 0)
616                         return r;
617                 else if (r > 0)
618                         continue;
619
620                 r = parse_address_key(p, "port", &port);
621                 if (r < 0)
622                         return r;
623                 else if (r > 0)
624                         continue;
625
626                 r = parse_address_key(p, "family", &family);
627                 if (r < 0)
628                         return r;
629                 else if (r > 0)
630                         continue;
631
632                 skip_address_key(p);
633         }
634
635         if (!host || !port)
636                 return -EINVAL;
637
638         if (family) {
639                 if (streq(family, "ipv4"))
640                         hints.ai_family = AF_INET;
641                 else if (streq(family, "ipv6"))
642                         hints.ai_family = AF_INET6;
643                 else
644                         return -EINVAL;
645         }
646
647         r = getaddrinfo(host, port, &hints, &result);
648         if (r == EAI_SYSTEM)
649                 return -errno;
650         else if (r != 0)
651                 return -EADDRNOTAVAIL;
652
653         memcpy(&b->sockaddr, result->ai_addr, result->ai_addrlen);
654         b->sockaddr_size = result->ai_addrlen;
655
656         freeaddrinfo(result);
657
658         return 0;
659 }
660
661 static int parse_exec_address(sd_bus *b, const char **p, char **guid) {
662         char *path = NULL;
663         unsigned n_argv = 0, j;
664         char **argv = NULL;
665         size_t allocated = 0;
666         int r;
667
668         assert(b);
669         assert(p);
670         assert(*p);
671         assert(guid);
672
673         while (**p != 0 && **p != ';') {
674                 r = parse_address_key(p, "guid", guid);
675                 if (r < 0)
676                         goto fail;
677                 else if (r > 0)
678                         continue;
679
680                 r = parse_address_key(p, "path", &path);
681                 if (r < 0)
682                         goto fail;
683                 else if (r > 0)
684                         continue;
685
686                 if (startswith(*p, "argv")) {
687                         unsigned ul;
688
689                         errno = 0;
690                         ul = strtoul(*p + 4, (char**) p, 10);
691                         if (errno > 0 || **p != '=' || ul > 256) {
692                                 r = -EINVAL;
693                                 goto fail;
694                         }
695
696                         (*p) ++;
697
698                         if (ul >= n_argv) {
699                                 if (!GREEDY_REALLOC0(argv, allocated, ul + 2)) {
700                                         r = -ENOMEM;
701                                         goto fail;
702                                 }
703
704                                 n_argv = ul + 1;
705                         }
706
707                         r = parse_address_key(p, NULL, argv + ul);
708                         if (r < 0)
709                                 goto fail;
710
711                         continue;
712                 }
713
714                 skip_address_key(p);
715         }
716
717         if (!path) {
718                 r = -EINVAL;
719                 goto fail;
720         }
721
722         /* Make sure there are no holes in the array, with the
723          * exception of argv[0] */
724         for (j = 1; j < n_argv; j++)
725                 if (!argv[j]) {
726                         r = -EINVAL;
727                         goto fail;
728                 }
729
730         if (argv && argv[0] == NULL) {
731                 argv[0] = strdup(path);
732                 if (!argv[0]) {
733                         r = -ENOMEM;
734                         goto fail;
735                 }
736         }
737
738         b->exec_path = path;
739         b->exec_argv = argv;
740         return 0;
741
742 fail:
743         for (j = 0; j < n_argv; j++)
744                 free(argv[j]);
745
746         free(argv);
747         free(path);
748         return r;
749 }
750
751 static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
752         _cleanup_free_ char *path = NULL;
753         int r;
754
755         assert(b);
756         assert(p);
757         assert(*p);
758         assert(guid);
759
760         while (**p != 0 && **p != ';') {
761                 r = parse_address_key(p, "guid", guid);
762                 if (r < 0)
763                         return r;
764                 else if (r > 0)
765                         continue;
766
767                 r = parse_address_key(p, "path", &path);
768                 if (r < 0)
769                         return r;
770                 else if (r > 0)
771                         continue;
772
773                 skip_address_key(p);
774         }
775
776         if (!path)
777                 return -EINVAL;
778
779         free(b->kernel);
780         b->kernel = path;
781         path = NULL;
782
783         return 0;
784 }
785
786 static int parse_container_unix_address(sd_bus *b, const char **p, char **guid) {
787         _cleanup_free_ char *machine = NULL, *pid = NULL;
788         int r;
789
790         assert(b);
791         assert(p);
792         assert(*p);
793         assert(guid);
794
795         while (**p != 0 && **p != ';') {
796                 r = parse_address_key(p, "guid", guid);
797                 if (r < 0)
798                         return r;
799                 else if (r > 0)
800                         continue;
801
802                 r = parse_address_key(p, "machine", &machine);
803                 if (r < 0)
804                         return r;
805                 else if (r > 0)
806                         continue;
807
808                 r = parse_address_key(p, "pid", &pid);
809                 if (r < 0)
810                         return r;
811                 else if (r > 0)
812                         continue;
813
814                 skip_address_key(p);
815         }
816
817         if (!machine == !pid)
818                 return -EINVAL;
819
820         if (machine) {
821                 if (!machine_name_is_valid(machine))
822                         return -EINVAL;
823
824                 free(b->machine);
825                 b->machine = machine;
826                 machine = NULL;
827         } else {
828                 free(b->machine);
829                 b->machine = NULL;
830         }
831
832         if (pid) {
833                 r = parse_pid(pid, &b->nspid);
834                 if (r < 0)
835                         return r;
836         } else
837                 b->nspid = 0;
838
839         b->sockaddr.un.sun_family = AF_UNIX;
840         strncpy(b->sockaddr.un.sun_path, "/var/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
841         b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + strlen("/var/run/dbus/system_bus_socket");
842
843         return 0;
844 }
845
846 static int parse_container_kernel_address(sd_bus *b, const char **p, char **guid) {
847         _cleanup_free_ char *machine = NULL, *pid = NULL;
848         int r;
849
850         assert(b);
851         assert(p);
852         assert(*p);
853         assert(guid);
854
855         while (**p != 0 && **p != ';') {
856                 r = parse_address_key(p, "guid", guid);
857                 if (r < 0)
858                         return r;
859                 else if (r > 0)
860                         continue;
861
862                 r = parse_address_key(p, "machine", &machine);
863                 if (r < 0)
864                         return r;
865                 else if (r > 0)
866                         continue;
867
868                 r = parse_address_key(p, "pid", &pid);
869                 if (r < 0)
870                         return r;
871                 else if (r > 0)
872                         continue;
873
874                 skip_address_key(p);
875         }
876
877         if (!machine == !pid)
878                 return -EINVAL;
879
880         if (machine) {
881                 if (!machine_name_is_valid(machine))
882                         return -EINVAL;
883
884                 free(b->machine);
885                 b->machine = machine;
886                 machine = NULL;
887         } else {
888                 free(b->machine);
889                 b->machine = NULL;
890         }
891
892         if (pid) {
893                 r = parse_pid(pid, &b->nspid);
894                 if (r < 0)
895                         return r;
896         } else
897                 b->nspid = 0;
898
899         free(b->kernel);
900         b->kernel = strdup("/sys/fs/kdbus/0-system/bus");
901         if (!b->kernel)
902                 return -ENOMEM;
903
904         return 0;
905 }
906
907 static void bus_reset_parsed_address(sd_bus *b) {
908         assert(b);
909
910         zero(b->sockaddr);
911         b->sockaddr_size = 0;
912         strv_free(b->exec_argv);
913         free(b->exec_path);
914         b->exec_path = NULL;
915         b->exec_argv = NULL;
916         b->server_id = SD_ID128_NULL;
917         free(b->kernel);
918         b->kernel = NULL;
919         free(b->machine);
920         b->machine = NULL;
921         b->nspid = 0;
922 }
923
924 static int bus_parse_next_address(sd_bus *b) {
925         _cleanup_free_ char *guid = NULL;
926         const char *a;
927         int r;
928
929         assert(b);
930
931         if (!b->address)
932                 return 0;
933         if (b->address[b->address_index] == 0)
934                 return 0;
935
936         bus_reset_parsed_address(b);
937
938         a = b->address + b->address_index;
939
940         while (*a != 0) {
941
942                 if (*a == ';') {
943                         a++;
944                         continue;
945                 }
946
947                 if (startswith(a, "unix:")) {
948                         a += 5;
949
950                         r = parse_unix_address(b, &a, &guid);
951                         if (r < 0)
952                                 return r;
953                         break;
954
955                 } else if (startswith(a, "tcp:")) {
956
957                         a += 4;
958                         r = parse_tcp_address(b, &a, &guid);
959                         if (r < 0)
960                                 return r;
961
962                         break;
963
964                 } else if (startswith(a, "unixexec:")) {
965
966                         a += 9;
967                         r = parse_exec_address(b, &a, &guid);
968                         if (r < 0)
969                                 return r;
970
971                         break;
972
973                 } else if (startswith(a, "kernel:")) {
974
975                         a += 7;
976                         r = parse_kernel_address(b, &a, &guid);
977                         if (r < 0)
978                                 return r;
979
980                         break;
981                 } else if (startswith(a, "x-machine-unix:")) {
982
983                         a += 15;
984                         r = parse_container_unix_address(b, &a, &guid);
985                         if (r < 0)
986                                 return r;
987
988                         break;
989                 } else if (startswith(a, "x-machine-kernel:")) {
990
991                         a += 17;
992                         r = parse_container_kernel_address(b, &a, &guid);
993                         if (r < 0)
994                                 return r;
995
996                         break;
997                 }
998
999                 a = strchr(a, ';');
1000                 if (!a)
1001                         return 0;
1002         }
1003
1004         if (guid) {
1005                 r = sd_id128_from_string(guid, &b->server_id);
1006                 if (r < 0)
1007                         return r;
1008         }
1009
1010         b->address_index = a - b->address;
1011         return 1;
1012 }
1013
1014 static int bus_start_address(sd_bus *b) {
1015         int r;
1016
1017         assert(b);
1018
1019         for (;;) {
1020                 bool skipped = false;
1021
1022                 bus_close_fds(b);
1023
1024                 if (b->exec_path)
1025                         r = bus_socket_exec(b);
1026                 else if ((b->nspid > 0 || b->machine) && b->kernel)
1027                         r = bus_container_connect_kernel(b);
1028                 else if ((b->nspid > 0 || b->machine) && b->sockaddr.sa.sa_family != AF_UNSPEC)
1029                         r = bus_container_connect_socket(b);
1030                 else if (b->kernel)
1031                         r = bus_kernel_connect(b);
1032                 else if (b->sockaddr.sa.sa_family != AF_UNSPEC)
1033                         r = bus_socket_connect(b);
1034                 else
1035                         skipped = true;
1036
1037                 if (!skipped) {
1038                         if (r >= 0) {
1039                                 r = attach_io_events(b);
1040                                 if (r >= 0)
1041                                         return r;
1042                         }
1043
1044                         b->last_connect_error = -r;
1045                 }
1046
1047                 r = bus_parse_next_address(b);
1048                 if (r < 0)
1049                         return r;
1050                 if (r == 0)
1051                         return b->last_connect_error ? -b->last_connect_error : -ECONNREFUSED;
1052         }
1053 }
1054
1055 int bus_next_address(sd_bus *b) {
1056         assert(b);
1057
1058         bus_reset_parsed_address(b);
1059         return bus_start_address(b);
1060 }
1061
1062 static int bus_start_fd(sd_bus *b) {
1063         struct stat st;
1064         int r;
1065
1066         assert(b);
1067         assert(b->input_fd >= 0);
1068         assert(b->output_fd >= 0);
1069
1070         r = fd_nonblock(b->input_fd, true);
1071         if (r < 0)
1072                 return r;
1073
1074         r = fd_cloexec(b->input_fd, true);
1075         if (r < 0)
1076                 return r;
1077
1078         if (b->input_fd != b->output_fd) {
1079                 r = fd_nonblock(b->output_fd, true);
1080                 if (r < 0)
1081                         return r;
1082
1083                 r = fd_cloexec(b->output_fd, true);
1084                 if (r < 0)
1085                         return r;
1086         }
1087
1088         if (fstat(b->input_fd, &st) < 0)
1089                 return -errno;
1090
1091         if (S_ISCHR(b->input_fd))
1092                 return bus_kernel_take_fd(b);
1093         else
1094                 return bus_socket_take_fd(b);
1095 }
1096
1097 _public_ int sd_bus_start(sd_bus *bus) {
1098         int r;
1099
1100         assert_return(bus, -EINVAL);
1101         assert_return(bus->state == BUS_UNSET, -EPERM);
1102         assert_return(!bus_pid_changed(bus), -ECHILD);
1103
1104         bus->state = BUS_OPENING;
1105
1106         if (bus->is_server && bus->bus_client)
1107                 return -EINVAL;
1108
1109         if (bus->input_fd >= 0)
1110                 r = bus_start_fd(bus);
1111         else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel || bus->machine)
1112                 r = bus_start_address(bus);
1113         else
1114                 return -EINVAL;
1115
1116         if (r < 0) {
1117                 sd_bus_close(bus);
1118                 return r;
1119         }
1120
1121         return bus_send_hello(bus);
1122 }
1123
1124 _public_ int sd_bus_open(sd_bus **ret) {
1125         const char *e;
1126         sd_bus *b;
1127         int r;
1128
1129         assert_return(ret, -EINVAL);
1130
1131         /* Let's connect to the starter bus if it is set, and
1132          * otherwise to the bus that is appropropriate for the scope
1133          * we are running in */
1134
1135         e = secure_getenv("DBUS_STARTER_BUS_TYPE");
1136         if (e) {
1137                 if (streq(e, "system"))
1138                         return sd_bus_open_system(ret);
1139                 else if (STR_IN_SET(e, "session", "user"))
1140                         return sd_bus_open_user(ret);
1141         }
1142
1143         e = secure_getenv("DBUS_STARTER_ADDRESS");
1144         if (!e) {
1145                 return sd_bus_open_system(ret);
1146         }
1147
1148         r = sd_bus_new(&b);
1149         if (r < 0)
1150                 return r;
1151
1152         r = sd_bus_set_address(b, e);
1153         if (r < 0)
1154                 goto fail;
1155
1156         b->bus_client = true;
1157
1158         /* We don't know whether the bus is trusted or not, so better
1159          * be safe, and authenticate everything */
1160         b->trusted = false;
1161         b->attach_flags |= KDBUS_ATTACH_CAPS | KDBUS_ATTACH_CREDS;
1162         b->creds_mask |= SD_BUS_CREDS_UID | SD_BUS_CREDS_EUID | SD_BUS_CREDS_EFFECTIVE_CAPS;
1163
1164         r = sd_bus_start(b);
1165         if (r < 0)
1166                 goto fail;
1167
1168         *ret = b;
1169         return 0;
1170
1171 fail:
1172         bus_free(b);
1173         return r;
1174 }
1175
1176 int bus_set_address_system(sd_bus *b) {
1177         const char *e;
1178         assert(b);
1179
1180         e = secure_getenv("DBUS_SYSTEM_BUS_ADDRESS");
1181         if (e)
1182                 return sd_bus_set_address(b, e);
1183
1184         return sd_bus_set_address(b, DEFAULT_SYSTEM_BUS_ADDRESS);
1185 }
1186
1187 _public_ int sd_bus_open_system(sd_bus **ret) {
1188         sd_bus *b;
1189         int r;
1190
1191         assert_return(ret, -EINVAL);
1192
1193         r = sd_bus_new(&b);
1194         if (r < 0)
1195                 return r;
1196
1197         r = bus_set_address_system(b);
1198         if (r < 0)
1199                 goto fail;
1200
1201         b->bus_client = true;
1202         b->is_system = true;
1203
1204         /* Let's do per-method access control on the system bus. We
1205          * need the caller's UID and capability set for that. */
1206         b->trusted = false;
1207         b->attach_flags |= KDBUS_ATTACH_CAPS | KDBUS_ATTACH_CREDS;
1208         b->creds_mask |= SD_BUS_CREDS_UID | SD_BUS_CREDS_EUID | SD_BUS_CREDS_EFFECTIVE_CAPS;
1209
1210         r = sd_bus_start(b);
1211         if (r < 0)
1212                 goto fail;
1213
1214         *ret = b;
1215         return 0;
1216
1217 fail:
1218         bus_free(b);
1219         return r;
1220 }
1221
1222 int bus_set_address_user(sd_bus *b) {
1223         const char *e;
1224
1225         assert(b);
1226
1227         e = secure_getenv("DBUS_SESSION_BUS_ADDRESS");
1228         if (e)
1229                 return sd_bus_set_address(b, e);
1230
1231         e = secure_getenv("XDG_RUNTIME_DIR");
1232         if (e) {
1233                 _cleanup_free_ char *ee = NULL;
1234
1235                 ee = bus_address_escape(e);
1236                 if (!ee)
1237                         return -ENOMEM;
1238
1239 #ifdef ENABLE_KDBUS
1240                 (void) asprintf(&b->address, KERNEL_USER_BUS_ADDRESS_FMT ";" UNIX_USER_BUS_ADDRESS_FMT, getuid(), ee);
1241 #else
1242                 (void) asprintf(&b->address, UNIX_USER_BUS_ADDRESS_FMT, ee);
1243 #endif
1244         } else {
1245 #ifdef ENABLE_KDBUS
1246                 (void) asprintf(&b->address, KERNEL_USER_BUS_ADDRESS_FMT, getuid());
1247 #else
1248                 return -ECONNREFUSED;
1249 #endif
1250         }
1251
1252         if (!b->address)
1253                 return -ENOMEM;
1254
1255         return 0;
1256 }
1257
1258 _public_ int sd_bus_open_user(sd_bus **ret) {
1259         sd_bus *b;
1260         int r;
1261
1262         assert_return(ret, -EINVAL);
1263
1264         r = sd_bus_new(&b);
1265         if (r < 0)
1266                 return r;
1267
1268         r = bus_set_address_user(b);
1269         if (r < 0)
1270                 return r;
1271
1272         b->bus_client = true;
1273         b->is_user = true;
1274
1275         /* We don't do any per-method access control on the user
1276          * bus. */
1277         b->trusted = true;
1278
1279         r = sd_bus_start(b);
1280         if (r < 0)
1281                 goto fail;
1282
1283         *ret = b;
1284         return 0;
1285
1286 fail:
1287         bus_free(b);
1288         return r;
1289 }
1290
1291 int bus_set_address_system_remote(sd_bus *b, const char *host) {
1292         _cleanup_free_ char *e = NULL;
1293         char *m = NULL, *c = NULL;
1294
1295         assert(b);
1296         assert(host);
1297
1298         /* Let's see if we shall enter some container */
1299         m = strchr(host, ':');
1300         if (m) {
1301                 m++;
1302
1303                 /* Let's make sure this is not a port of some kind,
1304                  * and is a valid machine name. */
1305                 if (!in_charset(m, "0123456789") && machine_name_is_valid(m)) {
1306                         char *t;
1307
1308                         /* Cut out the host part */
1309                         t = strndupa(host, m - host - 1);
1310                         e = bus_address_escape(t);
1311                         if (!e)
1312                                 return -ENOMEM;
1313
1314                         c = strjoina(",argv4=--machine=", m);
1315                 }
1316         }
1317
1318         if (!e) {
1319                 e = bus_address_escape(host);
1320                 if (!e)
1321                         return -ENOMEM;
1322         }
1323
1324         b->address = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", c, NULL);
1325         if (!b->address)
1326                 return -ENOMEM;
1327
1328         return 0;
1329  }
1330
1331 _public_ int sd_bus_open_system_remote(sd_bus **ret, const char *host) {
1332         sd_bus *bus;
1333         int r;
1334
1335         assert_return(host, -EINVAL);
1336         assert_return(ret, -EINVAL);
1337
1338         r = sd_bus_new(&bus);
1339         if (r < 0)
1340                 return r;
1341
1342         r = bus_set_address_system_remote(bus, host);
1343         if (r < 0)
1344                 goto fail;
1345
1346         bus->bus_client = true;
1347         bus->trusted = false;
1348         bus->is_system = true;
1349
1350         r = sd_bus_start(bus);
1351         if (r < 0)
1352                 goto fail;
1353
1354         *ret = bus;
1355         return 0;
1356
1357 fail:
1358         bus_free(bus);
1359         return r;
1360 }
1361
1362 int bus_set_address_system_machine(sd_bus *b, const char *machine) {
1363         _cleanup_free_ char *e = NULL;
1364
1365         assert(b);
1366         assert(machine);
1367
1368         e = bus_address_escape(machine);
1369         if (!e)
1370                 return -ENOMEM;
1371
1372 #ifdef ENABLE_KDBUS
1373         b->address = strjoin("x-machine-kernel:machine=", e, ";x-machine-unix:machine=", e, NULL);
1374 #else
1375         b->address = strjoin("x-machine-unix:machine=", e, NULL);
1376 #endif
1377         if (!b->address)
1378                 return -ENOMEM;
1379
1380         return 0;
1381 }
1382
1383 _public_ int sd_bus_open_system_machine(sd_bus **ret, const char *machine) {
1384         sd_bus *bus;
1385         int r;
1386
1387         assert_return(machine, -EINVAL);
1388         assert_return(ret, -EINVAL);
1389         assert_return(machine_name_is_valid(machine), -EINVAL);
1390
1391         r = sd_bus_new(&bus);
1392         if (r < 0)
1393                 return r;
1394
1395         r = bus_set_address_system_machine(bus, machine);
1396         if (r < 0)
1397                 goto fail;
1398
1399         bus->bus_client = true;
1400         bus->trusted = false;
1401         bus->is_system = true;
1402
1403         r = sd_bus_start(bus);
1404         if (r < 0)
1405                 goto fail;
1406
1407         *ret = bus;
1408         return 0;
1409
1410 fail:
1411         bus_free(bus);
1412         return r;
1413 }
1414
1415 _public_ void sd_bus_close(sd_bus *bus) {
1416
1417         if (!bus)
1418                 return;
1419         if (bus->state == BUS_CLOSED)
1420                 return;
1421         if (bus_pid_changed(bus))
1422                 return;
1423
1424         bus->state = BUS_CLOSED;
1425
1426         sd_bus_detach_event(bus);
1427
1428         /* Drop all queued messages so that they drop references to
1429          * the bus object and the bus may be freed */
1430         bus_reset_queues(bus);
1431
1432         if (!bus->is_kernel)
1433                 bus_close_fds(bus);
1434
1435         /* We'll leave the fd open in case this is a kernel bus, since
1436          * there might still be memblocks around that reference this
1437          * bus, and they might need to invoke the KDBUS_CMD_FREE
1438          * ioctl on the fd when they are freed. */
1439 }
1440
1441 static void bus_enter_closing(sd_bus *bus) {
1442         assert(bus);
1443
1444         if (bus->state != BUS_OPENING &&
1445             bus->state != BUS_AUTHENTICATING &&
1446             bus->state != BUS_HELLO &&
1447             bus->state != BUS_RUNNING)
1448                 return;
1449
1450         bus->state = BUS_CLOSING;
1451 }
1452
1453 _public_ sd_bus *sd_bus_ref(sd_bus *bus) {
1454         assert_return(bus, NULL);
1455
1456         assert_se(REFCNT_INC(bus->n_ref) >= 2);
1457
1458         return bus;
1459 }
1460
1461 _public_ sd_bus *sd_bus_unref(sd_bus *bus) {
1462         unsigned i;
1463
1464         if (!bus)
1465                 return NULL;
1466
1467         i = REFCNT_DEC(bus->n_ref);
1468         if (i > 0)
1469                 return NULL;
1470
1471         bus_free(bus);
1472         return NULL;
1473 }
1474
1475 _public_ int sd_bus_is_open(sd_bus *bus) {
1476
1477         assert_return(bus, -EINVAL);
1478         assert_return(!bus_pid_changed(bus), -ECHILD);
1479
1480         return BUS_IS_OPEN(bus->state);
1481 }
1482
1483 _public_ int sd_bus_can_send(sd_bus *bus, char type) {
1484         int r;
1485
1486         assert_return(bus, -EINVAL);
1487         assert_return(bus->state != BUS_UNSET, -ENOTCONN);
1488         assert_return(!bus_pid_changed(bus), -ECHILD);
1489
1490         if (bus->hello_flags & KDBUS_HELLO_MONITOR)
1491                 return 0;
1492
1493         if (type == SD_BUS_TYPE_UNIX_FD) {
1494                 if (!(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD))
1495                         return 0;
1496
1497                 r = bus_ensure_running(bus);
1498                 if (r < 0)
1499                         return r;
1500
1501                 return bus->can_fds;
1502         }
1503
1504         return bus_type_is_valid(type);
1505 }
1506
1507 _public_ int sd_bus_get_bus_id(sd_bus *bus, sd_id128_t *id) {
1508         int r;
1509
1510         assert_return(bus, -EINVAL);
1511         assert_return(id, -EINVAL);
1512         assert_return(!bus_pid_changed(bus), -ECHILD);
1513
1514         r = bus_ensure_running(bus);
1515         if (r < 0)
1516                 return r;
1517
1518         *id = bus->server_id;
1519         return 0;
1520 }
1521
1522 static int bus_seal_message(sd_bus *b, sd_bus_message *m, usec_t timeout) {
1523         assert(b);
1524         assert(m);
1525
1526         if (m->sealed) {
1527                 /* If we copy the same message to multiple
1528                  * destinations, avoid using the same cookie
1529                  * numbers. */
1530                 b->cookie = MAX(b->cookie, BUS_MESSAGE_COOKIE(m));
1531                 return 0;
1532         }
1533
1534         if (timeout == 0)
1535                 timeout = BUS_DEFAULT_TIMEOUT;
1536
1537         return bus_message_seal(m, ++b->cookie, timeout);
1538 }
1539
1540 static int bus_remarshal_message(sd_bus *b, sd_bus_message **m) {
1541         bool remarshal = false;
1542
1543         assert(b);
1544
1545         /* wrong packet version */
1546         if (b->message_version != 0 && b->message_version != (*m)->header->version)
1547                 remarshal = true;
1548
1549         /* wrong packet endianness */
1550         if (b->message_endian != 0 && b->message_endian != (*m)->header->endian)
1551                 remarshal = true;
1552
1553         /* TODO: kdbus-messages received from the kernel contain data which is
1554          * not allowed to be passed to KDBUS_CMD_SEND. Therefore, we have to
1555          * force remarshaling of the message. Technically, we could just
1556          * recreate the kdbus message, but that is non-trivial as other parts of
1557          * the message refer to m->kdbus already. This should be fixed! */
1558         if ((*m)->kdbus && (*m)->release_kdbus)
1559                 remarshal = true;
1560
1561         return remarshal ? bus_message_remarshal(b, m) : 0;
1562 }
1563
1564 int bus_seal_synthetic_message(sd_bus *b, sd_bus_message *m) {
1565         assert(b);
1566         assert(m);
1567
1568         /* Fake some timestamps, if they were requested, and not
1569          * already initialized */
1570         if (b->attach_flags & KDBUS_ATTACH_TIMESTAMP) {
1571                 if (m->realtime <= 0)
1572                         m->realtime = now(CLOCK_REALTIME);
1573
1574                 if (m->monotonic <= 0)
1575                         m->monotonic = now(CLOCK_MONOTONIC);
1576         }
1577
1578         /* The bus specification says the serial number cannot be 0,
1579          * hence let's fill something in for synthetic messages. Since
1580          * synthetic messages might have a fake sender and we don't
1581          * want to interfere with the real sender's serial numbers we
1582          * pick a fixed, artificial one. We use (uint32_t) -1 rather
1583          * than (uint64_t) -1 since dbus1 only had 32bit identifiers,
1584          * even though kdbus can do 64bit. */
1585         return bus_message_seal(m, 0xFFFFFFFFULL, 0);
1586 }
1587
1588 static int bus_write_message(sd_bus *bus, sd_bus_message *m, bool hint_sync_call, size_t *idx) {
1589         int r;
1590
1591         assert(bus);
1592         assert(m);
1593
1594         if (bus->is_kernel)
1595                 r = bus_kernel_write_message(bus, m, hint_sync_call);
1596         else
1597                 r = bus_socket_write_message(bus, m, idx);
1598
1599         if (r <= 0)
1600                 return r;
1601
1602         if (bus->is_kernel || *idx >= BUS_MESSAGE_SIZE(m))
1603                 log_debug("Sent message type=%s sender=%s destination=%s object=%s interface=%s member=%s cookie=%" PRIu64 " reply_cookie=%" PRIu64 " error=%s",
1604                           bus_message_type_to_string(m->header->type),
1605                           strna(sd_bus_message_get_sender(m)),
1606                           strna(sd_bus_message_get_destination(m)),
1607                           strna(sd_bus_message_get_path(m)),
1608                           strna(sd_bus_message_get_interface(m)),
1609                           strna(sd_bus_message_get_member(m)),
1610                           BUS_MESSAGE_COOKIE(m),
1611                           m->reply_cookie,
1612                           strna(m->error.message));
1613
1614         return r;
1615 }
1616
1617 static int dispatch_wqueue(sd_bus *bus) {
1618         int r, ret = 0;
1619
1620         assert(bus);
1621         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1622
1623         while (bus->wqueue_size > 0) {
1624
1625                 r = bus_write_message(bus, bus->wqueue[0], false, &bus->windex);
1626                 if (r < 0)
1627                         return r;
1628                 else if (r == 0)
1629                         /* Didn't do anything this time */
1630                         return ret;
1631                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1632                         /* Fully written. Let's drop the entry from
1633                          * the queue.
1634                          *
1635                          * This isn't particularly optimized, but
1636                          * well, this is supposed to be our worst-case
1637                          * buffer only, and the socket buffer is
1638                          * supposed to be our primary buffer, and if
1639                          * it got full, then all bets are off
1640                          * anyway. */
1641
1642                         bus->wqueue_size --;
1643                         sd_bus_message_unref(bus->wqueue[0]);
1644                         memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1645                         bus->windex = 0;
1646
1647                         ret = 1;
1648                 }
1649         }
1650
1651         return ret;
1652 }
1653
1654 static int bus_read_message(sd_bus *bus, bool hint_priority, int64_t priority) {
1655         assert(bus);
1656
1657         if (bus->is_kernel)
1658                 return bus_kernel_read_message(bus, hint_priority, priority);
1659         else
1660                 return bus_socket_read_message(bus);
1661 }
1662
1663 int bus_rqueue_make_room(sd_bus *bus) {
1664         assert(bus);
1665
1666         if (bus->rqueue_size >= BUS_RQUEUE_MAX)
1667                 return -ENOBUFS;
1668
1669         if (!GREEDY_REALLOC(bus->rqueue, bus->rqueue_allocated, bus->rqueue_size + 1))
1670                 return -ENOMEM;
1671
1672         return 0;
1673 }
1674
1675 static int dispatch_rqueue(sd_bus *bus, bool hint_priority, int64_t priority, sd_bus_message **m) {
1676         int r, ret = 0;
1677
1678         assert(bus);
1679         assert(m);
1680         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1681
1682         /* Note that the priority logic is only available on kdbus,
1683          * where the rqueue is unused. We check the rqueue here
1684          * anyway, because it's simple... */
1685
1686         for (;;) {
1687                 if (bus->rqueue_size > 0) {
1688                         /* Dispatch a queued message */
1689
1690                         *m = bus->rqueue[0];
1691                         bus->rqueue_size --;
1692                         memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1693                         return 1;
1694                 }
1695
1696                 /* Try to read a new message */
1697                 r = bus_read_message(bus, hint_priority, priority);
1698                 if (r < 0)
1699                         return r;
1700                 if (r == 0)
1701                         return ret;
1702
1703                 ret = 1;
1704         }
1705 }
1706
1707 static int bus_send_internal(sd_bus *bus, sd_bus_message *_m, uint64_t *cookie, bool hint_sync_call) {
1708         _cleanup_bus_message_unref_ sd_bus_message *m = sd_bus_message_ref(_m);
1709         int r;
1710
1711         assert_return(m, -EINVAL);
1712
1713         if (!bus)
1714                 bus = m->bus;
1715
1716         assert_return(!bus_pid_changed(bus), -ECHILD);
1717         assert_return(!bus->is_kernel || !(bus->hello_flags & KDBUS_HELLO_MONITOR), -EROFS);
1718
1719         if (!BUS_IS_OPEN(bus->state))
1720                 return -ENOTCONN;
1721
1722         if (m->n_fds > 0) {
1723                 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1724                 if (r < 0)
1725                         return r;
1726                 if (r == 0)
1727                         return -EOPNOTSUPP;
1728         }
1729
1730         /* If the cookie number isn't kept, then we know that no reply
1731          * is expected */
1732         if (!cookie && !m->sealed)
1733                 m->header->flags |= BUS_MESSAGE_NO_REPLY_EXPECTED;
1734
1735         r = bus_seal_message(bus, m, 0);
1736         if (r < 0)
1737                 return r;
1738
1739         /* Remarshall if we have to. This will possibly unref the
1740          * message and place a replacement in m */
1741         r = bus_remarshal_message(bus, &m);
1742         if (r < 0)
1743                 return r;
1744
1745         /* If this is a reply and no reply was requested, then let's
1746          * suppress this, if we can */
1747         if (m->dont_send)
1748                 goto finish;
1749
1750         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1751                 size_t idx = 0;
1752
1753                 r = bus_write_message(bus, m, hint_sync_call, &idx);
1754                 if (r < 0) {
1755                         if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
1756                                 bus_enter_closing(bus);
1757                                 return -ECONNRESET;
1758                         }
1759
1760                         return r;
1761                 }
1762
1763                 if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
1764                         /* Wasn't fully written. So let's remember how
1765                          * much was written. Note that the first entry
1766                          * of the wqueue array is always allocated so
1767                          * that we always can remember how much was
1768                          * written. */
1769                         bus->wqueue[0] = sd_bus_message_ref(m);
1770                         bus->wqueue_size = 1;
1771                         bus->windex = idx;
1772                 }
1773
1774         } else {
1775                 /* Just append it to the queue. */
1776
1777                 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1778                         return -ENOBUFS;
1779
1780                 if (!GREEDY_REALLOC(bus->wqueue, bus->wqueue_allocated, bus->wqueue_size + 1))
1781                         return -ENOMEM;
1782
1783                 bus->wqueue[bus->wqueue_size ++] = sd_bus_message_ref(m);
1784         }
1785
1786 finish:
1787         if (cookie)
1788                 *cookie = BUS_MESSAGE_COOKIE(m);
1789
1790         return 1;
1791 }
1792
1793 _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *cookie) {
1794         return bus_send_internal(bus, m, cookie, false);
1795 }
1796
1797 _public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *cookie) {
1798         int r;
1799
1800         assert_return(m, -EINVAL);
1801
1802         if (!bus)
1803                 bus = m->bus;
1804
1805         assert_return(!bus_pid_changed(bus), -ECHILD);
1806
1807         if (!BUS_IS_OPEN(bus->state))
1808                 return -ENOTCONN;
1809
1810         if (!streq_ptr(m->destination, destination)) {
1811
1812                 if (!destination)
1813                         return -EEXIST;
1814
1815                 r = sd_bus_message_set_destination(m, destination);
1816                 if (r < 0)
1817                         return r;
1818         }
1819
1820         return sd_bus_send(bus, m, cookie);
1821 }
1822
1823 static usec_t calc_elapse(uint64_t usec) {
1824         if (usec == (uint64_t) -1)
1825                 return 0;
1826
1827         return now(CLOCK_MONOTONIC) + usec;
1828 }
1829
1830 static int timeout_compare(const void *a, const void *b) {
1831         const struct reply_callback *x = a, *y = b;
1832
1833         if (x->timeout != 0 && y->timeout == 0)
1834                 return -1;
1835
1836         if (x->timeout == 0 && y->timeout != 0)
1837                 return 1;
1838
1839         if (x->timeout < y->timeout)
1840                 return -1;
1841
1842         if (x->timeout > y->timeout)
1843                 return 1;
1844
1845         return 0;
1846 }
1847
1848 _public_ int sd_bus_call_async(
1849                 sd_bus *bus,
1850                 sd_bus_slot **slot,
1851                 sd_bus_message *_m,
1852                 sd_bus_message_handler_t callback,
1853                 void *userdata,
1854                 uint64_t usec) {
1855
1856         _cleanup_bus_message_unref_ sd_bus_message *m = sd_bus_message_ref(_m);
1857         _cleanup_bus_slot_unref_ sd_bus_slot *s = NULL;
1858         int r;
1859
1860         assert_return(m, -EINVAL);
1861         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1862         assert_return(!(m->header->flags & BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1863         assert_return(callback, -EINVAL);
1864
1865         if (!bus)
1866                 bus = m->bus;
1867
1868         assert_return(!bus_pid_changed(bus), -ECHILD);
1869         assert_return(!bus->is_kernel || !(bus->hello_flags & KDBUS_HELLO_MONITOR), -EROFS);
1870
1871         if (!BUS_IS_OPEN(bus->state))
1872                 return -ENOTCONN;
1873
1874         r = ordered_hashmap_ensure_allocated(&bus->reply_callbacks, &uint64_hash_ops);
1875         if (r < 0)
1876                 return r;
1877
1878         r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1879         if (r < 0)
1880                 return r;
1881
1882         r = bus_seal_message(bus, m, usec);
1883         if (r < 0)
1884                 return r;
1885
1886         r = bus_remarshal_message(bus, &m);
1887         if (r < 0)
1888                 return r;
1889
1890         s = bus_slot_allocate(bus, !slot, BUS_REPLY_CALLBACK, sizeof(struct reply_callback), userdata);
1891         if (!s)
1892                 return -ENOMEM;
1893
1894         s->reply_callback.callback = callback;
1895
1896         s->reply_callback.cookie = BUS_MESSAGE_COOKIE(m);
1897         r = ordered_hashmap_put(bus->reply_callbacks, &s->reply_callback.cookie, &s->reply_callback);
1898         if (r < 0) {
1899                 s->reply_callback.cookie = 0;
1900                 return r;
1901         }
1902
1903         s->reply_callback.timeout = calc_elapse(m->timeout);
1904         if (s->reply_callback.timeout != 0) {
1905                 r = prioq_put(bus->reply_callbacks_prioq, &s->reply_callback, &s->reply_callback.prioq_idx);
1906                 if (r < 0) {
1907                         s->reply_callback.timeout = 0;
1908                         return r;
1909                 }
1910         }
1911
1912         r = sd_bus_send(bus, m, &s->reply_callback.cookie);
1913         if (r < 0)
1914                 return r;
1915
1916         if (slot)
1917                 *slot = s;
1918         s = NULL;
1919
1920         return r;
1921 }
1922
1923 int bus_ensure_running(sd_bus *bus) {
1924         int r;
1925
1926         assert(bus);
1927
1928         if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
1929                 return -ENOTCONN;
1930         if (bus->state == BUS_RUNNING)
1931                 return 1;
1932
1933         for (;;) {
1934                 r = sd_bus_process(bus, NULL);
1935                 if (r < 0)
1936                         return r;
1937                 if (bus->state == BUS_RUNNING)
1938                         return 1;
1939                 if (r > 0)
1940                         continue;
1941
1942                 r = sd_bus_wait(bus, (uint64_t) -1);
1943                 if (r < 0)
1944                         return r;
1945         }
1946 }
1947
1948 _public_ int sd_bus_call(
1949                 sd_bus *bus,
1950                 sd_bus_message *_m,
1951                 uint64_t usec,
1952                 sd_bus_error *error,
1953                 sd_bus_message **reply) {
1954
1955         _cleanup_bus_message_unref_ sd_bus_message *m = sd_bus_message_ref(_m);
1956         usec_t timeout;
1957         uint64_t cookie;
1958         unsigned i;
1959         int r;
1960
1961         assert_return(m, -EINVAL);
1962         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1963         assert_return(!(m->header->flags & BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1964         assert_return(!bus_error_is_dirty(error), -EINVAL);
1965
1966         if (!bus)
1967                 bus = m->bus;
1968
1969         assert_return(!bus_pid_changed(bus), -ECHILD);
1970         assert_return(!bus->is_kernel || !(bus->hello_flags & KDBUS_HELLO_MONITOR), -EROFS);
1971
1972         if (!BUS_IS_OPEN(bus->state))
1973                 return -ENOTCONN;
1974
1975         r = bus_ensure_running(bus);
1976         if (r < 0)
1977                 return r;
1978
1979         i = bus->rqueue_size;
1980
1981         r = bus_seal_message(bus, m, usec);
1982         if (r < 0)
1983                 return r;
1984
1985         r = bus_remarshal_message(bus, &m);
1986         if (r < 0)
1987                 return r;
1988
1989         r = bus_send_internal(bus, m, &cookie, true);
1990         if (r < 0)
1991                 return r;
1992
1993         timeout = calc_elapse(m->timeout);
1994
1995         for (;;) {
1996                 usec_t left;
1997
1998                 while (i < bus->rqueue_size) {
1999                         sd_bus_message *incoming = NULL;
2000
2001                         incoming = bus->rqueue[i];
2002
2003                         if (incoming->reply_cookie == cookie) {
2004                                 /* Found a match! */
2005
2006                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
2007                                 bus->rqueue_size--;
2008                                 log_debug_bus_message(incoming);
2009
2010                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
2011
2012                                         if (incoming->n_fds <= 0 || (bus->hello_flags & KDBUS_HELLO_ACCEPT_FD)) {
2013                                                 if (reply)
2014                                                         *reply = incoming;
2015                                                 else
2016                                                         sd_bus_message_unref(incoming);
2017
2018                                                 return 1;
2019                                         }
2020
2021                                         r = sd_bus_error_setf(error, SD_BUS_ERROR_INCONSISTENT_MESSAGE, "Reply message contained file descriptors which I couldn't accept. Sorry.");
2022
2023                                 } else if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR)
2024                                         r = sd_bus_error_copy(error, &incoming->error);
2025                                 else
2026                                         r = -EIO;
2027
2028                                 sd_bus_message_unref(incoming);
2029                                 return r;
2030
2031                         } else if (BUS_MESSAGE_COOKIE(incoming) == cookie &&
2032                                    bus->unique_name &&
2033                                    incoming->sender &&
2034                                    streq(bus->unique_name, incoming->sender)) {
2035
2036                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
2037                                 bus->rqueue_size--;
2038
2039                                 /* Our own message? Somebody is trying
2040                                  * to send its own client a message,
2041                                  * let's not dead-lock, let's fail
2042                                  * immediately. */
2043
2044                                 sd_bus_message_unref(incoming);
2045                                 return -ELOOP;
2046                         }
2047
2048                         /* Try to read more, right-away */
2049                         i++;
2050                 }
2051
2052                 r = bus_read_message(bus, false, 0);
2053                 if (r < 0) {
2054                         if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2055                                 bus_enter_closing(bus);
2056                                 return -ECONNRESET;
2057                         }
2058
2059                         return r;
2060                 }
2061                 if (r > 0)
2062                         continue;
2063
2064                 if (timeout > 0) {
2065                         usec_t n;
2066
2067                         n = now(CLOCK_MONOTONIC);
2068                         if (n >= timeout)
2069                                 return -ETIMEDOUT;
2070
2071                         left = timeout - n;
2072                 } else
2073                         left = (uint64_t) -1;
2074
2075                 r = bus_poll(bus, true, left);
2076                 if (r < 0)
2077                         return r;
2078                 if (r == 0)
2079                         return -ETIMEDOUT;
2080
2081                 r = dispatch_wqueue(bus);
2082                 if (r < 0) {
2083                         if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2084                                 bus_enter_closing(bus);
2085                                 return -ECONNRESET;
2086                         }
2087
2088                         return r;
2089                 }
2090         }
2091 }
2092
2093 _public_ int sd_bus_get_fd(sd_bus *bus) {
2094
2095         assert_return(bus, -EINVAL);
2096         assert_return(bus->input_fd == bus->output_fd, -EPERM);
2097         assert_return(!bus_pid_changed(bus), -ECHILD);
2098
2099         return bus->input_fd;
2100 }
2101
2102 _public_ int sd_bus_get_events(sd_bus *bus) {
2103         int flags = 0;
2104
2105         assert_return(bus, -EINVAL);
2106         assert_return(!bus_pid_changed(bus), -ECHILD);
2107
2108         if (!BUS_IS_OPEN(bus->state) && bus->state != BUS_CLOSING)
2109                 return -ENOTCONN;
2110
2111         if (bus->state == BUS_OPENING)
2112                 flags |= POLLOUT;
2113         else if (bus->state == BUS_AUTHENTICATING) {
2114
2115                 if (bus_socket_auth_needs_write(bus))
2116                         flags |= POLLOUT;
2117
2118                 flags |= POLLIN;
2119
2120         } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
2121                 if (bus->rqueue_size <= 0)
2122                         flags |= POLLIN;
2123                 if (bus->wqueue_size > 0)
2124                         flags |= POLLOUT;
2125         }
2126
2127         return flags;
2128 }
2129
2130 _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
2131         struct reply_callback *c;
2132
2133         assert_return(bus, -EINVAL);
2134         assert_return(timeout_usec, -EINVAL);
2135         assert_return(!bus_pid_changed(bus), -ECHILD);
2136
2137         if (!BUS_IS_OPEN(bus->state) && bus->state != BUS_CLOSING)
2138                 return -ENOTCONN;
2139
2140         if (bus->track_queue) {
2141                 *timeout_usec = 0;
2142                 return 1;
2143         }
2144
2145         if (bus->state == BUS_CLOSING) {
2146                 *timeout_usec = 0;
2147                 return 1;
2148         }
2149
2150         if (bus->state == BUS_AUTHENTICATING) {
2151                 *timeout_usec = bus->auth_timeout;
2152                 return 1;
2153         }
2154
2155         if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
2156                 *timeout_usec = (uint64_t) -1;
2157                 return 0;
2158         }
2159
2160         if (bus->rqueue_size > 0) {
2161                 *timeout_usec = 0;
2162                 return 1;
2163         }
2164
2165         c = prioq_peek(bus->reply_callbacks_prioq);
2166         if (!c) {
2167                 *timeout_usec = (uint64_t) -1;
2168                 return 0;
2169         }
2170
2171         if (c->timeout == 0) {
2172                 *timeout_usec = (uint64_t) -1;
2173                 return 0;
2174         }
2175
2176         *timeout_usec = c->timeout;
2177         return 1;
2178 }
2179
2180 static int process_timeout(sd_bus *bus) {
2181         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2182         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
2183         struct reply_callback *c;
2184         sd_bus_slot *slot;
2185         usec_t n;
2186         int r;
2187
2188         assert(bus);
2189
2190         c = prioq_peek(bus->reply_callbacks_prioq);
2191         if (!c)
2192                 return 0;
2193
2194         n = now(CLOCK_MONOTONIC);
2195         if (c->timeout > n)
2196                 return 0;
2197
2198         r = bus_message_new_synthetic_error(
2199                         bus,
2200                         c->cookie,
2201                         &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
2202                         &m);
2203         if (r < 0)
2204                 return r;
2205
2206         r = bus_seal_synthetic_message(bus, m);
2207         if (r < 0)
2208                 return r;
2209
2210         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
2211         c->timeout = 0;
2212
2213         ordered_hashmap_remove(bus->reply_callbacks, &c->cookie);
2214         c->cookie = 0;
2215
2216         slot = container_of(c, sd_bus_slot, reply_callback);
2217
2218         bus->iteration_counter ++;
2219
2220         bus->current_message = m;
2221         bus->current_slot = sd_bus_slot_ref(slot);
2222         bus->current_handler = c->callback;
2223         bus->current_userdata = slot->userdata;
2224         r = c->callback(m, slot->userdata, &error_buffer);
2225         bus->current_userdata = NULL;
2226         bus->current_handler = NULL;
2227         bus->current_slot = NULL;
2228         bus->current_message = NULL;
2229
2230         if (slot->floating) {
2231                 bus_slot_disconnect(slot);
2232                 sd_bus_slot_unref(slot);
2233         }
2234
2235         sd_bus_slot_unref(slot);
2236
2237         return bus_maybe_reply_error(m, r, &error_buffer);
2238 }
2239
2240 static int process_hello(sd_bus *bus, sd_bus_message *m) {
2241         assert(bus);
2242         assert(m);
2243
2244         if (bus->state != BUS_HELLO)
2245                 return 0;
2246
2247         /* Let's make sure the first message on the bus is the HELLO
2248          * reply. But note that we don't actually parse the message
2249          * here (we leave that to the usual handling), we just verify
2250          * we don't let any earlier msg through. */
2251
2252         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
2253             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
2254                 return -EIO;
2255
2256         if (m->reply_cookie != 1)
2257                 return -EIO;
2258
2259         return 0;
2260 }
2261
2262 static int process_reply(sd_bus *bus, sd_bus_message *m) {
2263         _cleanup_bus_message_unref_ sd_bus_message *synthetic_reply = NULL;
2264         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2265         struct reply_callback *c;
2266         sd_bus_slot *slot;
2267         int r;
2268
2269         assert(bus);
2270         assert(m);
2271
2272         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
2273             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
2274                 return 0;
2275
2276         if (bus->is_kernel && (bus->hello_flags & KDBUS_HELLO_MONITOR))
2277                 return 0;
2278
2279         if (m->destination && bus->unique_name && !streq_ptr(m->destination, bus->unique_name))
2280                 return 0;
2281
2282         c = ordered_hashmap_remove(bus->reply_callbacks, &m->reply_cookie);
2283         if (!c)
2284                 return 0;
2285
2286         c->cookie = 0;
2287
2288         slot = container_of(c, sd_bus_slot, reply_callback);
2289
2290         if (m->n_fds > 0 && !(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD)) {
2291
2292                 /* If the reply contained a file descriptor which we
2293                  * didn't want we pass an error instead. */
2294
2295                 r = bus_message_new_synthetic_error(
2296                                 bus,
2297                                 m->reply_cookie,
2298                                 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_INCONSISTENT_MESSAGE, "Reply message contained file descriptor"),
2299                                 &synthetic_reply);
2300                 if (r < 0)
2301                         return r;
2302
2303                 /* Copy over original timestamp */
2304                 synthetic_reply->realtime = m->realtime;
2305                 synthetic_reply->monotonic = m->monotonic;
2306                 synthetic_reply->seqnum = m->seqnum;
2307
2308                 r = bus_seal_synthetic_message(bus, synthetic_reply);
2309                 if (r < 0)
2310                         return r;
2311
2312                 m = synthetic_reply;
2313         } else {
2314                 r = sd_bus_message_rewind(m, true);
2315                 if (r < 0)
2316                         return r;
2317         }
2318
2319         if (c->timeout != 0) {
2320                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
2321                 c->timeout = 0;
2322         }
2323
2324         bus->current_slot = sd_bus_slot_ref(slot);
2325         bus->current_handler = c->callback;
2326         bus->current_userdata = slot->userdata;
2327         r = c->callback(m, slot->userdata, &error_buffer);
2328         bus->current_userdata = NULL;
2329         bus->current_handler = NULL;
2330         bus->current_slot = NULL;
2331
2332         if (slot->floating) {
2333                 bus_slot_disconnect(slot);
2334                 sd_bus_slot_unref(slot);
2335         }
2336
2337         sd_bus_slot_unref(slot);
2338
2339         return bus_maybe_reply_error(m, r, &error_buffer);
2340 }
2341
2342 static int process_filter(sd_bus *bus, sd_bus_message *m) {
2343         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2344         struct filter_callback *l;
2345         int r;
2346
2347         assert(bus);
2348         assert(m);
2349
2350         do {
2351                 bus->filter_callbacks_modified = false;
2352
2353                 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
2354                         sd_bus_slot *slot;
2355
2356                         if (bus->filter_callbacks_modified)
2357                                 break;
2358
2359                         /* Don't run this more than once per iteration */
2360                         if (l->last_iteration == bus->iteration_counter)
2361                                 continue;
2362
2363                         l->last_iteration = bus->iteration_counter;
2364
2365                         r = sd_bus_message_rewind(m, true);
2366                         if (r < 0)
2367                                 return r;
2368
2369                         slot = container_of(l, sd_bus_slot, filter_callback);
2370
2371                         bus->current_slot = sd_bus_slot_ref(slot);
2372                         bus->current_handler = l->callback;
2373                         bus->current_userdata = slot->userdata;
2374                         r = l->callback(m, slot->userdata, &error_buffer);
2375                         bus->current_userdata = NULL;
2376                         bus->current_handler = NULL;
2377                         bus->current_slot = sd_bus_slot_unref(slot);
2378
2379                         r = bus_maybe_reply_error(m, r, &error_buffer);
2380                         if (r != 0)
2381                                 return r;
2382
2383                 }
2384
2385         } while (bus->filter_callbacks_modified);
2386
2387         return 0;
2388 }
2389
2390 static int process_match(sd_bus *bus, sd_bus_message *m) {
2391         int r;
2392
2393         assert(bus);
2394         assert(m);
2395
2396         do {
2397                 bus->match_callbacks_modified = false;
2398
2399                 r = bus_match_run(bus, &bus->match_callbacks, m);
2400                 if (r != 0)
2401                         return r;
2402
2403         } while (bus->match_callbacks_modified);
2404
2405         return 0;
2406 }
2407
2408 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
2409         _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
2410         int r;
2411
2412         assert(bus);
2413         assert(m);
2414
2415         if (bus->hello_flags & KDBUS_HELLO_MONITOR)
2416                 return 0;
2417
2418         if (bus->manual_peer_interface)
2419                 return 0;
2420
2421         if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
2422                 return 0;
2423
2424         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
2425                 return 0;
2426
2427         if (m->header->flags & BUS_MESSAGE_NO_REPLY_EXPECTED)
2428                 return 1;
2429
2430         if (streq_ptr(m->member, "Ping"))
2431                 r = sd_bus_message_new_method_return(m, &reply);
2432         else if (streq_ptr(m->member, "GetMachineId")) {
2433                 sd_id128_t id;
2434                 char sid[33];
2435
2436                 r = sd_id128_get_machine(&id);
2437                 if (r < 0)
2438                         return r;
2439
2440                 r = sd_bus_message_new_method_return(m, &reply);
2441                 if (r < 0)
2442                         return r;
2443
2444                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
2445         } else {
2446                 r = sd_bus_message_new_method_errorf(
2447                                 m, &reply,
2448                                 SD_BUS_ERROR_UNKNOWN_METHOD,
2449                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
2450         }
2451
2452         if (r < 0)
2453                 return r;
2454
2455         r = sd_bus_send(bus, reply, NULL);
2456         if (r < 0)
2457                 return r;
2458
2459         return 1;
2460 }
2461
2462 static int process_fd_check(sd_bus *bus, sd_bus_message *m) {
2463         assert(bus);
2464         assert(m);
2465
2466         /* If we got a message with a file descriptor which we didn't
2467          * want to accept, then let's drop it. How can this even
2468          * happen? For example, when the kernel queues a message into
2469          * an activatable names's queue which allows fds, and then is
2470          * delivered to us later even though we ourselves did not
2471          * negotiate it. */
2472
2473         if (bus->hello_flags & KDBUS_HELLO_MONITOR)
2474                 return 0;
2475
2476         if (m->n_fds <= 0)
2477                 return 0;
2478
2479         if (bus->hello_flags & KDBUS_HELLO_ACCEPT_FD)
2480                 return 0;
2481
2482         if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
2483                 return 1; /* just eat it up */
2484
2485         return sd_bus_reply_method_errorf(m, SD_BUS_ERROR_INCONSISTENT_MESSAGE, "Message contains file descriptors, which I cannot accept. Sorry.");
2486 }
2487
2488 static int process_message(sd_bus *bus, sd_bus_message *m) {
2489         int r;
2490
2491         assert(bus);
2492         assert(m);
2493
2494         bus->current_message = m;
2495         bus->iteration_counter++;
2496
2497         log_debug_bus_message(m);
2498
2499         r = process_hello(bus, m);
2500         if (r != 0)
2501                 goto finish;
2502
2503         r = process_reply(bus, m);
2504         if (r != 0)
2505                 goto finish;
2506
2507         r = process_fd_check(bus, m);
2508         if (r != 0)
2509                 goto finish;
2510
2511         r = process_filter(bus, m);
2512         if (r != 0)
2513                 goto finish;
2514
2515         r = process_match(bus, m);
2516         if (r != 0)
2517                 goto finish;
2518
2519         r = process_builtin(bus, m);
2520         if (r != 0)
2521                 goto finish;
2522
2523         r = bus_process_object(bus, m);
2524
2525 finish:
2526         bus->current_message = NULL;
2527         return r;
2528 }
2529
2530 static int dispatch_track(sd_bus *bus) {
2531         assert(bus);
2532
2533         if (!bus->track_queue)
2534                 return 0;
2535
2536         bus_track_dispatch(bus->track_queue);
2537         return 1;
2538 }
2539
2540 static int process_running(sd_bus *bus, bool hint_priority, int64_t priority, sd_bus_message **ret) {
2541         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2542         int r;
2543
2544         assert(bus);
2545         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
2546
2547         r = process_timeout(bus);
2548         if (r != 0)
2549                 goto null_message;
2550
2551         r = dispatch_wqueue(bus);
2552         if (r != 0)
2553                 goto null_message;
2554
2555         r = dispatch_track(bus);
2556         if (r != 0)
2557                 goto null_message;
2558
2559         r = dispatch_rqueue(bus, hint_priority, priority, &m);
2560         if (r < 0)
2561                 return r;
2562         if (!m)
2563                 goto null_message;
2564
2565         r = process_message(bus, m);
2566         if (r != 0)
2567                 goto null_message;
2568
2569         if (ret) {
2570                 r = sd_bus_message_rewind(m, true);
2571                 if (r < 0)
2572                         return r;
2573
2574                 *ret = m;
2575                 m = NULL;
2576                 return 1;
2577         }
2578
2579         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
2580
2581                 log_debug("Unprocessed message call sender=%s object=%s interface=%s member=%s",
2582                           strna(sd_bus_message_get_sender(m)),
2583                           strna(sd_bus_message_get_path(m)),
2584                           strna(sd_bus_message_get_interface(m)),
2585                           strna(sd_bus_message_get_member(m)));
2586
2587                 r = sd_bus_reply_method_errorf(
2588                                 m,
2589                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
2590                                 "Unknown object '%s'.", m->path);
2591                 if (r < 0)
2592                         return r;
2593         }
2594
2595         return 1;
2596
2597 null_message:
2598         if (r >= 0 && ret)
2599                 *ret = NULL;
2600
2601         return r;
2602 }
2603
2604 static int process_closing(sd_bus *bus, sd_bus_message **ret) {
2605         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2606         struct reply_callback *c;
2607         int r;
2608
2609         assert(bus);
2610         assert(bus->state == BUS_CLOSING);
2611
2612         c = ordered_hashmap_first(bus->reply_callbacks);
2613         if (c) {
2614                 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2615                 sd_bus_slot *slot;
2616
2617                 /* First, fail all outstanding method calls */
2618                 r = bus_message_new_synthetic_error(
2619                                 bus,
2620                                 c->cookie,
2621                                 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
2622                                 &m);
2623                 if (r < 0)
2624                         return r;
2625
2626                 r = bus_seal_synthetic_message(bus, m);
2627                 if (r < 0)
2628                         return r;
2629
2630                 if (c->timeout != 0) {
2631                         prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
2632                         c->timeout = 0;
2633                 }
2634
2635                 ordered_hashmap_remove(bus->reply_callbacks, &c->cookie);
2636                 c->cookie = 0;
2637
2638                 slot = container_of(c, sd_bus_slot, reply_callback);
2639
2640                 bus->iteration_counter++;
2641
2642                 bus->current_message = m;
2643                 bus->current_slot = sd_bus_slot_ref(slot);
2644                 bus->current_handler = c->callback;
2645                 bus->current_userdata = slot->userdata;
2646                 r = c->callback(m, slot->userdata, &error_buffer);
2647                 bus->current_userdata = NULL;
2648                 bus->current_handler = NULL;
2649                 bus->current_slot = NULL;
2650                 bus->current_message = NULL;
2651
2652                 if (slot->floating) {
2653                         bus_slot_disconnect(slot);
2654                         sd_bus_slot_unref(slot);
2655                 }
2656
2657                 sd_bus_slot_unref(slot);
2658
2659                 return bus_maybe_reply_error(m, r, &error_buffer);
2660         }
2661
2662         /* Then, synthesize a Disconnected message */
2663         r = sd_bus_message_new_signal(
2664                         bus,
2665                         &m,
2666                         "/org/freedesktop/DBus/Local",
2667                         "org.freedesktop.DBus.Local",
2668                         "Disconnected");
2669         if (r < 0)
2670                 return r;
2671
2672         bus_message_set_sender_local(bus, m);
2673
2674         r = bus_seal_synthetic_message(bus, m);
2675         if (r < 0)
2676                 return r;
2677
2678         sd_bus_close(bus);
2679
2680         bus->current_message = m;
2681         bus->iteration_counter++;
2682
2683         r = process_filter(bus, m);
2684         if (r != 0)
2685                 goto finish;
2686
2687         r = process_match(bus, m);
2688         if (r != 0)
2689                 goto finish;
2690
2691         if (ret) {
2692                 *ret = m;
2693                 m = NULL;
2694         }
2695
2696         r = 1;
2697
2698 finish:
2699         bus->current_message = NULL;
2700
2701         return r;
2702 }
2703
2704 static int bus_process_internal(sd_bus *bus, bool hint_priority, int64_t priority, sd_bus_message **ret) {
2705         BUS_DONT_DESTROY(bus);
2706         int r;
2707
2708         /* Returns 0 when we didn't do anything. This should cause the
2709          * caller to invoke sd_bus_wait() before returning the next
2710          * time. Returns > 0 when we did something, which possibly
2711          * means *ret is filled in with an unprocessed message. */
2712
2713         assert_return(bus, -EINVAL);
2714         assert_return(!bus_pid_changed(bus), -ECHILD);
2715
2716         /* We don't allow recursively invoking sd_bus_process(). */
2717         assert_return(!bus->current_message, -EBUSY);
2718         assert(!bus->current_slot);
2719
2720         switch (bus->state) {
2721
2722         case BUS_UNSET:
2723                 return -ENOTCONN;
2724
2725         case BUS_CLOSED:
2726                 return -ECONNRESET;
2727
2728         case BUS_OPENING:
2729                 r = bus_socket_process_opening(bus);
2730                 if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2731                         bus_enter_closing(bus);
2732                         r = 1;
2733                 } else if (r < 0)
2734                         return r;
2735                 if (ret)
2736                         *ret = NULL;
2737                 return r;
2738
2739         case BUS_AUTHENTICATING:
2740                 r = bus_socket_process_authenticating(bus);
2741                 if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2742                         bus_enter_closing(bus);
2743                         r = 1;
2744                 } else if (r < 0)
2745                         return r;
2746
2747                 if (ret)
2748                         *ret = NULL;
2749
2750                 return r;
2751
2752         case BUS_RUNNING:
2753         case BUS_HELLO:
2754                 r = process_running(bus, hint_priority, priority, ret);
2755                 if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2756                         bus_enter_closing(bus);
2757                         r = 1;
2758
2759                         if (ret)
2760                                 *ret = NULL;
2761                 }
2762
2763                 return r;
2764
2765         case BUS_CLOSING:
2766                 return process_closing(bus, ret);
2767         }
2768
2769         assert_not_reached("Unknown state");
2770 }
2771
2772 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
2773         return bus_process_internal(bus, false, 0, ret);
2774 }
2775
2776 _public_ int sd_bus_process_priority(sd_bus *bus, int64_t priority, sd_bus_message **ret) {
2777         return bus_process_internal(bus, true, priority, ret);
2778 }
2779
2780 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
2781         struct pollfd p[2] = {};
2782         int r, e, n;
2783         struct timespec ts;
2784         usec_t m = USEC_INFINITY;
2785
2786         assert(bus);
2787
2788         if (bus->state == BUS_CLOSING)
2789                 return 1;
2790
2791         if (!BUS_IS_OPEN(bus->state))
2792                 return -ENOTCONN;
2793
2794         e = sd_bus_get_events(bus);
2795         if (e < 0)
2796                 return e;
2797
2798         if (need_more)
2799                 /* The caller really needs some more data, he doesn't
2800                  * care about what's already read, or any timeouts
2801                  * except its own. */
2802                 e |= POLLIN;
2803         else {
2804                 usec_t until;
2805                 /* The caller wants to process if there's something to
2806                  * process, but doesn't care otherwise */
2807
2808                 r = sd_bus_get_timeout(bus, &until);
2809                 if (r < 0)
2810                         return r;
2811                 if (r > 0) {
2812                         usec_t nw;
2813                         nw = now(CLOCK_MONOTONIC);
2814                         m = until > nw ? until - nw : 0;
2815                 }
2816         }
2817
2818         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
2819                 m = timeout_usec;
2820
2821         p[0].fd = bus->input_fd;
2822         if (bus->output_fd == bus->input_fd) {
2823                 p[0].events = e;
2824                 n = 1;
2825         } else {
2826                 p[0].events = e & POLLIN;
2827                 p[1].fd = bus->output_fd;
2828                 p[1].events = e & POLLOUT;
2829                 n = 2;
2830         }
2831
2832         r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
2833         if (r < 0)
2834                 return -errno;
2835
2836         return r > 0 ? 1 : 0;
2837 }
2838
2839 _public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
2840
2841         assert_return(bus, -EINVAL);
2842         assert_return(!bus_pid_changed(bus), -ECHILD);
2843
2844         if (bus->state == BUS_CLOSING)
2845                 return 0;
2846
2847         if (!BUS_IS_OPEN(bus->state))
2848                 return -ENOTCONN;
2849
2850         if (bus->rqueue_size > 0)
2851                 return 0;
2852
2853         return bus_poll(bus, false, timeout_usec);
2854 }
2855
2856 _public_ int sd_bus_flush(sd_bus *bus) {
2857         int r;
2858
2859         assert_return(bus, -EINVAL);
2860         assert_return(!bus_pid_changed(bus), -ECHILD);
2861
2862         if (bus->state == BUS_CLOSING)
2863                 return 0;
2864
2865         if (!BUS_IS_OPEN(bus->state))
2866                 return -ENOTCONN;
2867
2868         r = bus_ensure_running(bus);
2869         if (r < 0)
2870                 return r;
2871
2872         if (bus->wqueue_size <= 0)
2873                 return 0;
2874
2875         for (;;) {
2876                 r = dispatch_wqueue(bus);
2877                 if (r < 0) {
2878                         if (r == -ENOTCONN || r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2879                                 bus_enter_closing(bus);
2880                                 return -ECONNRESET;
2881                         }
2882
2883                         return r;
2884                 }
2885
2886                 if (bus->wqueue_size <= 0)
2887                         return 0;
2888
2889                 r = bus_poll(bus, false, (uint64_t) -1);
2890                 if (r < 0)
2891                         return r;
2892         }
2893 }
2894
2895 _public_ int sd_bus_add_filter(
2896                 sd_bus *bus,
2897                 sd_bus_slot **slot,
2898                 sd_bus_message_handler_t callback,
2899                 void *userdata) {
2900
2901         sd_bus_slot *s;
2902
2903         assert_return(bus, -EINVAL);
2904         assert_return(callback, -EINVAL);
2905         assert_return(!bus_pid_changed(bus), -ECHILD);
2906
2907         s = bus_slot_allocate(bus, !slot, BUS_FILTER_CALLBACK, sizeof(struct filter_callback), userdata);
2908         if (!s)
2909                 return -ENOMEM;
2910
2911         s->filter_callback.callback = callback;
2912
2913         bus->filter_callbacks_modified = true;
2914         LIST_PREPEND(callbacks, bus->filter_callbacks, &s->filter_callback);
2915
2916         if (slot)
2917                 *slot = s;
2918
2919         return 0;
2920 }
2921
2922 _public_ int sd_bus_add_match(
2923                 sd_bus *bus,
2924                 sd_bus_slot **slot,
2925                 const char *match,
2926                 sd_bus_message_handler_t callback,
2927                 void *userdata) {
2928
2929         struct bus_match_component *components = NULL;
2930         unsigned n_components = 0;
2931         sd_bus_slot *s = NULL;
2932         int r = 0;
2933
2934         assert_return(bus, -EINVAL);
2935         assert_return(match, -EINVAL);
2936         assert_return(!bus_pid_changed(bus), -ECHILD);
2937
2938         r = bus_match_parse(match, &components, &n_components);
2939         if (r < 0)
2940                 goto finish;
2941
2942         s = bus_slot_allocate(bus, !slot, BUS_MATCH_CALLBACK, sizeof(struct match_callback), userdata);
2943         if (!s) {
2944                 r = -ENOMEM;
2945                 goto finish;
2946         }
2947
2948         s->match_callback.callback = callback;
2949         s->match_callback.cookie = ++bus->match_cookie;
2950
2951         if (bus->bus_client) {
2952
2953                 if (!bus->is_kernel) {
2954                         /* When this is not a kernel transport, we
2955                          * store the original match string, so that we
2956                          * can use it to remove the match again */
2957
2958                         s->match_callback.match_string = strdup(match);
2959                         if (!s->match_callback.match_string) {
2960                                 r = -ENOMEM;
2961                                 goto finish;
2962                         }
2963                 }
2964
2965                 r = bus_add_match_internal(bus, s->match_callback.match_string, components, n_components, s->match_callback.cookie);
2966                 if (r < 0)
2967                         goto finish;
2968         }
2969
2970         bus->match_callbacks_modified = true;
2971         r = bus_match_add(&bus->match_callbacks, components, n_components, &s->match_callback);
2972         if (r < 0)
2973                 goto finish;
2974
2975         if (slot)
2976                 *slot = s;
2977         s = NULL;
2978
2979 finish:
2980         bus_match_parse_free(components, n_components);
2981         sd_bus_slot_unref(s);
2982
2983         return r;
2984 }
2985
2986 int bus_remove_match_by_string(
2987                 sd_bus *bus,
2988                 const char *match,
2989                 sd_bus_message_handler_t callback,
2990                 void *userdata) {
2991
2992         struct bus_match_component *components = NULL;
2993         unsigned n_components = 0;
2994         struct match_callback *c;
2995         int r = 0;
2996
2997         assert_return(bus, -EINVAL);
2998         assert_return(match, -EINVAL);
2999         assert_return(!bus_pid_changed(bus), -ECHILD);
3000
3001         r = bus_match_parse(match, &components, &n_components);
3002         if (r < 0)
3003                 goto finish;
3004
3005         r = bus_match_find(&bus->match_callbacks, components, n_components, NULL, NULL, &c);
3006         if (r <= 0)
3007                 goto finish;
3008
3009         sd_bus_slot_unref(container_of(c, sd_bus_slot, match_callback));
3010
3011 finish:
3012         bus_match_parse_free(components, n_components);
3013
3014         return r;
3015 }
3016
3017 bool bus_pid_changed(sd_bus *bus) {
3018         assert(bus);
3019
3020         /* We don't support people creating a bus connection and
3021          * keeping it around over a fork(). Let's complain. */
3022
3023         return bus->original_pid != getpid();
3024 }
3025
3026 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
3027         sd_bus *bus = userdata;
3028         int r;
3029
3030         assert(bus);
3031
3032         r = sd_bus_process(bus, NULL);
3033         if (r < 0)
3034                 return r;
3035
3036         return 1;
3037 }
3038
3039 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
3040         sd_bus *bus = userdata;
3041         int r;
3042
3043         assert(bus);
3044
3045         r = sd_bus_process(bus, NULL);
3046         if (r < 0)
3047                 return r;
3048
3049         return 1;
3050 }
3051
3052 static int prepare_callback(sd_event_source *s, void *userdata) {
3053         sd_bus *bus = userdata;
3054         int r, e;
3055         usec_t until;
3056
3057         assert(s);
3058         assert(bus);
3059
3060         e = sd_bus_get_events(bus);
3061         if (e < 0)
3062                 return e;
3063
3064         if (bus->output_fd != bus->input_fd) {
3065
3066                 r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
3067                 if (r < 0)
3068                         return r;
3069
3070                 r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
3071                 if (r < 0)
3072                         return r;
3073         } else {
3074                 r = sd_event_source_set_io_events(bus->input_io_event_source, e);
3075                 if (r < 0)
3076                         return r;
3077         }
3078
3079         r = sd_bus_get_timeout(bus, &until);
3080         if (r < 0)
3081                 return r;
3082         if (r > 0) {
3083                 int j;
3084
3085                 j = sd_event_source_set_time(bus->time_event_source, until);
3086                 if (j < 0)
3087                         return j;
3088         }
3089
3090         r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
3091         if (r < 0)
3092                 return r;
3093
3094         return 1;
3095 }
3096
3097 static int quit_callback(sd_event_source *event, void *userdata) {
3098         sd_bus *bus = userdata;
3099
3100         assert(event);
3101
3102         sd_bus_flush(bus);
3103         sd_bus_close(bus);
3104
3105         return 1;
3106 }
3107
3108 static int attach_io_events(sd_bus *bus) {
3109         int r;
3110
3111         assert(bus);
3112
3113         if (bus->input_fd < 0)
3114                 return 0;
3115
3116         if (!bus->event)
3117                 return 0;
3118
3119         if (!bus->input_io_event_source) {
3120                 r = sd_event_add_io(bus->event, &bus->input_io_event_source, bus->input_fd, 0, io_callback, bus);
3121                 if (r < 0)
3122                         return r;
3123
3124                 r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
3125                 if (r < 0)
3126                         return r;
3127
3128                 r = sd_event_source_set_priority(bus->input_io_event_source, bus->event_priority);
3129                 if (r < 0)
3130                         return r;
3131
3132                 r = sd_event_source_set_description(bus->input_io_event_source, "bus-input");
3133         } else
3134                 r = sd_event_source_set_io_fd(bus->input_io_event_source, bus->input_fd);
3135
3136         if (r < 0)
3137                 return r;
3138
3139         if (bus->output_fd != bus->input_fd) {
3140                 assert(bus->output_fd >= 0);
3141
3142                 if (!bus->output_io_event_source) {
3143                         r = sd_event_add_io(bus->event, &bus->output_io_event_source, bus->output_fd, 0, io_callback, bus);
3144                         if (r < 0)
3145                                 return r;
3146
3147                         r = sd_event_source_set_priority(bus->output_io_event_source, bus->event_priority);
3148                         if (r < 0)
3149                                 return r;
3150
3151                         r = sd_event_source_set_description(bus->input_io_event_source, "bus-output");
3152                 } else
3153                         r = sd_event_source_set_io_fd(bus->output_io_event_source, bus->output_fd);
3154
3155                 if (r < 0)
3156                         return r;
3157         }
3158
3159         return 0;
3160 }
3161
3162 static void detach_io_events(sd_bus *bus) {
3163         assert(bus);
3164
3165         if (bus->input_io_event_source) {
3166                 sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
3167                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
3168         }
3169
3170         if (bus->output_io_event_source) {
3171                 sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
3172                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
3173         }
3174 }
3175
3176 _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
3177         int r;
3178
3179         assert_return(bus, -EINVAL);
3180         assert_return(!bus->event, -EBUSY);
3181
3182         assert(!bus->input_io_event_source);
3183         assert(!bus->output_io_event_source);
3184         assert(!bus->time_event_source);
3185
3186         if (event)
3187                 bus->event = sd_event_ref(event);
3188         else  {
3189                 r = sd_event_default(&bus->event);
3190                 if (r < 0)
3191                         return r;
3192         }
3193
3194         bus->event_priority = priority;
3195
3196         r = sd_event_add_time(bus->event, &bus->time_event_source, CLOCK_MONOTONIC, 0, 0, time_callback, bus);
3197         if (r < 0)
3198                 goto fail;
3199
3200         r = sd_event_source_set_priority(bus->time_event_source, priority);
3201         if (r < 0)
3202                 goto fail;
3203
3204         r = sd_event_source_set_description(bus->time_event_source, "bus-time");
3205         if (r < 0)
3206                 goto fail;
3207
3208         r = sd_event_add_exit(bus->event, &bus->quit_event_source, quit_callback, bus);
3209         if (r < 0)
3210                 goto fail;
3211
3212         r = sd_event_source_set_description(bus->quit_event_source, "bus-exit");
3213         if (r < 0)
3214                 goto fail;
3215
3216         r = attach_io_events(bus);
3217         if (r < 0)
3218                 goto fail;
3219
3220         return 0;
3221
3222 fail:
3223         sd_bus_detach_event(bus);
3224         return r;
3225 }
3226
3227 _public_ int sd_bus_detach_event(sd_bus *bus) {
3228         assert_return(bus, -EINVAL);
3229
3230         if (!bus->event)
3231                 return 0;
3232
3233         detach_io_events(bus);
3234
3235         if (bus->time_event_source) {
3236                 sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
3237                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
3238         }
3239
3240         if (bus->quit_event_source) {
3241                 sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
3242                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
3243         }
3244
3245         bus->event = sd_event_unref(bus->event);
3246         return 1;
3247 }
3248
3249 _public_ sd_event* sd_bus_get_event(sd_bus *bus) {
3250         assert_return(bus, NULL);
3251
3252         return bus->event;
3253 }
3254
3255 _public_ sd_bus_message* sd_bus_get_current_message(sd_bus *bus) {
3256         assert_return(bus, NULL);
3257
3258         return bus->current_message;
3259 }
3260
3261 _public_ sd_bus_slot* sd_bus_get_current_slot(sd_bus *bus) {
3262         assert_return(bus, NULL);
3263
3264         return bus->current_slot;