src/libsystemd-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 <assert.h>
  24 #include <stdlib.h>
  25 #include <unistd.h>
  26 #include <netdb.h>
  27 #include <sys/poll.h>
  28 #include <byteswap.h>
  29 #include <sys/mman.h>
  30 #include <pthread.h>
  31
  32 #include "util.h"
  33 #include "macro.h"
  34 #include "strv.h"
  35 #include "set.h"
  36 #include "missing.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-introspect.h"
  46 #include "bus-signature.h"
  47 #include "bus-objects.h"
  48 #include "bus-util.h"
  49 #include "bus-container.h"
  50
  51 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
  52
  53 static void bus_close_fds(sd_bus *b) {
  54         assert(b);
  55
  56         if (b->input_fd >= 0)
  57                 close_nointr_nofail(b->input_fd);
  58
  59         if (b->output_fd >= 0 && b->output_fd != b->input_fd)
  60                 close_nointr_nofail(b->output_fd);
  61
  62         b->input_fd = b->output_fd = -1;
  63 }
  64
  65 static void bus_node_destroy(sd_bus *b, struct node *n) {
  66         struct node_callback *c;
  67         struct node_vtable *v;
  68         struct node_enumerator *e;
  69
  70         assert(b);
  71
  72         if (!n)
  73                 return;
  74
  75         while (n->child)
  76                 bus_node_destroy(b, n->child);
  77
  78         while ((c = n->callbacks)) {
  79                 LIST_REMOVE(callbacks, n->callbacks, c);
  80                 free(c);
  81         }
  82
  83         while ((v = n->vtables)) {
  84                 LIST_REMOVE(vtables, n->vtables, v);
  85                 free(v->interface);
  86                 free(v);
  87         }
  88
  89         while ((e = n->enumerators)) {
  90                 LIST_REMOVE(enumerators, n->enumerators, e);
  91                 free(e);
  92         }
  93
  94         if (n->parent)
  95                 LIST_REMOVE(siblings, n->parent->child, n);
  96
  97         assert_se(hashmap_remove(b->nodes, n->path) == n);
  98         free(n->path);
  99         free(n);
 100 }
 101
 102 static void bus_free(sd_bus *b) {
 103         struct filter_callback *f;
 104         struct node *n;
 105         unsigned i;
 106
 107         assert(b);
 108
 109         sd_bus_detach_event(b);
 110
 111         bus_close_fds(b);
 112
 113         if (b->kdbus_buffer)
 114                 munmap(b->kdbus_buffer, KDBUS_POOL_SIZE);
 115
 116         free(b->rbuffer);
 117         free(b->unique_name);
 118         free(b->auth_buffer);
 119         free(b->address);
 120         free(b->kernel);
 121         free(b->machine);
 122
 123         free(b->exec_path);
 124         strv_free(b->exec_argv);
 125
 126         close_many(b->fds, b->n_fds);
 127         free(b->fds);
 128
 129         for (i = 0; i < b->rqueue_size; i++)
 130                 sd_bus_message_unref(b->rqueue[i]);
 131         free(b->rqueue);
 132
 133         for (i = 0; i < b->wqueue_size; i++)
 134                 sd_bus_message_unref(b->wqueue[i]);
 135         free(b->wqueue);
 136
 137         hashmap_free_free(b->reply_callbacks);
 138         prioq_free(b->reply_callbacks_prioq);
 139
 140         while ((f = b->filter_callbacks)) {
 141                 LIST_REMOVE(callbacks, b->filter_callbacks, f);
 142                 free(f);
 143         }
 144
 145         bus_match_free(&b->match_callbacks);
 146
 147         hashmap_free_free(b->vtable_methods);
 148         hashmap_free_free(b->vtable_properties);
 149
 150         while ((n = hashmap_first(b->nodes)))
 151                 bus_node_destroy(b, n);
 152
 153         hashmap_free(b->nodes);
 154
 155         bus_kernel_flush_memfd(b);
 156
 157         assert_se(pthread_mutex_destroy(&b->memfd_cache_mutex) == 0);
 158
 159         free(b);
 160 }
 161
 162 int sd_bus_new(sd_bus **ret) {
 163         sd_bus *r;
 164
 165         assert_return(ret, -EINVAL);
 166
 167         r = new0(sd_bus, 1);
 168         if (!r)
 169                 return -ENOMEM;
 170
 171         r->n_ref = REFCNT_INIT;
 172         r->input_fd = r->output_fd = -1;
 173         r->message_version = 1;
 174         r->hello_flags |= KDBUS_HELLO_ACCEPT_FD;
 175         r->original_pid = getpid();
 176
 177         assert_se(pthread_mutex_init(&r->memfd_cache_mutex, NULL) == 0);
 178
 179         /* We guarantee that wqueue always has space for at least one
 180          * entry */
 181         r->wqueue = new(sd_bus_message*, 1);
 182         if (!r->wqueue) {
 183                 free(r);
 184                 return -ENOMEM;
 185         }
 186
 187         *ret = r;
 188         return 0;
 189 }
 190
 191 int sd_bus_set_address(sd_bus *bus, const char *address) {
 192         char *a;
 193
 194         assert_return(bus, -EINVAL);
 195         assert_return(bus->state == BUS_UNSET, -EPERM);
 196         assert_return(address, -EINVAL);
 197         assert_return(!bus_pid_changed(bus), -ECHILD);
 198
 199         a = strdup(address);
 200         if (!a)
 201                 return -ENOMEM;
 202
 203         free(bus->address);
 204         bus->address = a;
 205
 206         return 0;
 207 }
 208
 209 int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
 210         assert_return(bus, -EINVAL);
 211         assert_return(bus->state == BUS_UNSET, -EPERM);
 212         assert_return(input_fd >= 0, -EINVAL);
 213         assert_return(output_fd >= 0, -EINVAL);
 214         assert_return(!bus_pid_changed(bus), -ECHILD);
 215
 216         bus->input_fd = input_fd;
 217         bus->output_fd = output_fd;
 218         return 0;
 219 }
 220
 221 int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
 222         char *p, **a;
 223
 224         assert_return(bus, -EINVAL);
 225         assert_return(bus->state == BUS_UNSET, -EPERM);
 226         assert_return(path, -EINVAL);
 227         assert_return(!strv_isempty(argv), -EINVAL);
 228         assert_return(!bus_pid_changed(bus), -ECHILD);
 229
 230         p = strdup(path);
 231         if (!p)
 232                 return -ENOMEM;
 233
 234         a = strv_copy(argv);
 235         if (!a) {
 236                 free(p);
 237                 return -ENOMEM;
 238         }
 239
 240         free(bus->exec_path);
 241         strv_free(bus->exec_argv);
 242
 243         bus->exec_path = p;
 244         bus->exec_argv = a;
 245
 246         return 0;
 247 }
 248
 249 int sd_bus_set_bus_client(sd_bus *bus, int b) {
 250         assert_return(bus, -EINVAL);
 251         assert_return(bus->state == BUS_UNSET, -EPERM);
 252         assert_return(!bus_pid_changed(bus), -ECHILD);
 253
 254         bus->bus_client = !!b;
 255         return 0;
 256 }
 257
 258 int sd_bus_negotiate_fds(sd_bus *bus, int b) {
 259         assert_return(bus, -EINVAL);
 260         assert_return(bus->state == BUS_UNSET, -EPERM);
 261         assert_return(!bus_pid_changed(bus), -ECHILD);
 262
 263         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ACCEPT_FD, b);
 264         return 0;
 265 }
 266
 267 int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
 268         assert_return(bus, -EINVAL);
 269         assert_return(bus->state == BUS_UNSET, -EPERM);
 270         assert_return(!bus_pid_changed(bus), -ECHILD);
 271
 272         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_COMM, b);
 273         return 0;
 274 }
 275
 276 int sd_bus_negotiate_attach_exe(sd_bus *bus, int b) {
 277         assert_return(bus, -EINVAL);
 278         assert_return(bus->state == BUS_UNSET, -EPERM);
 279         assert_return(!bus_pid_changed(bus), -ECHILD);
 280
 281         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_EXE, b);
 282         return 0;
 283 }
 284
 285 int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
 286         assert_return(bus, -EINVAL);
 287         assert_return(bus->state == BUS_UNSET, -EPERM);
 288         assert_return(!bus_pid_changed(bus), -ECHILD);
 289
 290         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CMDLINE, b);
 291         return 0;
 292 }
 293
 294 int sd_bus_negotiate_attach_cgroup(sd_bus *bus, int b) {
 295         assert_return(bus, -EINVAL);
 296         assert_return(bus->state == BUS_UNSET, -EPERM);
 297         assert_return(!bus_pid_changed(bus), -ECHILD);
 298
 299         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CGROUP, b);
 300         return 0;
 301 }
 302
 303 int sd_bus_negotiate_attach_caps(sd_bus *bus, int b) {
 304         assert_return(bus, -EINVAL);
 305         assert_return(bus->state == BUS_UNSET, -EPERM);
 306         assert_return(!bus_pid_changed(bus), -ECHILD);
 307
 308         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CAPS, b);
 309         return 0;
 310 }
 311
 312 int sd_bus_negotiate_attach_selinux_context(sd_bus *bus, int b) {
 313         assert_return(bus, -EINVAL);
 314         assert_return(bus->state == BUS_UNSET, -EPERM);
 315         assert_return(!bus_pid_changed(bus), -ECHILD);
 316
 317         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_SECLABEL, b);
 318         return 0;
 319 }
 320
 321 int sd_bus_negotiate_attach_audit(sd_bus *bus, int b) {
 322         assert_return(bus, -EINVAL);
 323         assert_return(bus->state == BUS_UNSET, -EPERM);
 324         assert_return(!bus_pid_changed(bus), -ECHILD);
 325
 326         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_AUDIT, b);
 327         return 0;
 328 }
 329
 330 int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
 331         assert_return(bus, -EINVAL);
 332         assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
 333         assert_return(bus->state == BUS_UNSET, -EPERM);
 334         assert_return(!bus_pid_changed(bus), -ECHILD);
 335
 336         bus->is_server = !!b;
 337         bus->server_id = server_id;
 338         return 0;
 339 }
 340
 341 int sd_bus_set_anonymous(sd_bus *bus, int b) {
 342         assert_return(bus, -EINVAL);
 343         assert_return(bus->state == BUS_UNSET, -EPERM);
 344         assert_return(!bus_pid_changed(bus), -ECHILD);
 345
 346         bus->anonymous_auth = !!b;
 347         return 0;
 348 }
 349
 350 static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
 351         const char *s;
 352         int r;
 353
 354         assert(bus);
 355         assert(bus->state == BUS_HELLO);
 356         assert(reply);
 357
 358         r = sd_bus_message_get_errno(reply);
 359         if (r < 0)
 360                 return r;
 361         if (r > 0)
 362                 return -r;
 363
 364         r = sd_bus_message_read(reply, "s", &s);
 365         if (r < 0)
 366                 return r;
 367
 368         if (!service_name_is_valid(s) || s[0] != ':')
 369                 return -EBADMSG;
 370
 371         bus->unique_name = strdup(s);
 372         if (!bus->unique_name)
 373                 return -ENOMEM;
 374
 375         bus->state = BUS_RUNNING;
 376
 377         return 1;
 378 }
 379
 380 static int bus_send_hello(sd_bus *bus) {
 381         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
 382         int r;
 383
 384         assert(bus);
 385
 386         if (!bus->bus_client || bus->is_kernel)
 387                 return 0;
 388
 389         r = sd_bus_message_new_method_call(
 390                         bus,
 391                         "org.freedesktop.DBus",
 392                         "/",
 393                         "org.freedesktop.DBus",
 394                         "Hello",
 395                         &m);
 396         if (r < 0)
 397                 return r;
 398
 399         return sd_bus_send_with_reply(bus, m, hello_callback, NULL, 0, &bus->hello_serial);
 400 }
 401
 402 int bus_start_running(sd_bus *bus) {
 403         assert(bus);
 404
 405         if (bus->bus_client && !bus->is_kernel) {
 406                 bus->state = BUS_HELLO;
 407                 return 1;
 408         }
 409
 410         bus->state = BUS_RUNNING;
 411         return 1;
 412 }
 413
 414 static int parse_address_key(const char **p, const char *key, char **value) {
 415         size_t l, n = 0;
 416         const char *a;
 417         char *r = NULL;
 418
 419         assert(p);
 420         assert(*p);
 421         assert(value);
 422
 423         if (key) {
 424                 l = strlen(key);
 425                 if (strncmp(*p, key, l) != 0)
 426                         return 0;
 427
 428                 if ((*p)[l] != '=')
 429                         return 0;
 430
 431                 if (*value)
 432                         return -EINVAL;
 433
 434                 a = *p + l + 1;
 435         } else
 436                 a = *p;
 437
 438         while (*a != ';' && *a != ',' && *a != 0) {
 439                 char c, *t;
 440
 441                 if (*a == '%') {
 442                         int x, y;
 443
 444                         x = unhexchar(a[1]);
 445                         if (x < 0) {
 446                                 free(r);
 447                                 return x;
 448                         }
 449
 450                         y = unhexchar(a[2]);
 451                         if (y < 0) {
 452                                 free(r);
 453                                 return y;
 454                         }
 455
 456                         c = (char) ((x << 4) | y);
 457                         a += 3;
 458                 } else {
 459                         c = *a;
 460                         a++;
 461                 }
 462
 463                 t = realloc(r, n + 2);
 464                 if (!t) {
 465                         free(r);
 466                         return -ENOMEM;
 467                 }
 468
 469                 r = t;
 470                 r[n++] = c;
 471         }
 472
 473         if (!r) {
 474                 r = strdup("");
 475                 if (!r)
 476                         return -ENOMEM;
 477         } else
 478                 r[n] = 0;
 479
 480         if (*a == ',')
 481                 a++;
 482
 483         *p = a;
 484
 485         free(*value);
 486         *value = r;
 487
 488         return 1;
 489 }
 490
 491 static void skip_address_key(const char **p) {
 492         assert(p);
 493         assert(*p);
 494
 495         *p += strcspn(*p, ",");
 496
 497         if (**p == ',')
 498                 (*p) ++;
 499 }
 500
 501 static int parse_unix_address(sd_bus *b, const char **p, char **guid) {
 502         _cleanup_free_ char *path = NULL, *abstract = NULL;
 503         size_t l;
 504         int r;
 505
 506         assert(b);
 507         assert(p);
 508         assert(*p);
 509         assert(guid);
 510
 511         while (**p != 0 && **p != ';') {
 512                 r = parse_address_key(p, "guid", guid);
 513                 if (r < 0)
 514                         return r;
 515                 else if (r > 0)
 516                         continue;
 517
 518                 r = parse_address_key(p, "path", &path);
 519                 if (r < 0)
 520                         return r;
 521                 else if (r > 0)
 522                         continue;
 523
 524                 r = parse_address_key(p, "abstract", &abstract);
 525                 if (r < 0)
 526                         return r;
 527                 else if (r > 0)
 528                         continue;
 529
 530                 skip_address_key(p);
 531         }
 532
 533         if (!path && !abstract)
 534                 return -EINVAL;
 535
 536         if (path && abstract)
 537                 return -EINVAL;
 538
 539         if (path) {
 540                 l = strlen(path);
 541                 if (l > sizeof(b->sockaddr.un.sun_path))
 542                         return -E2BIG;
 543
 544                 b->sockaddr.un.sun_family = AF_UNIX;
 545                 strncpy(b->sockaddr.un.sun_path, path, sizeof(b->sockaddr.un.sun_path));
 546                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l;
 547         } else if (abstract) {
 548                 l = strlen(abstract);
 549                 if (l > sizeof(b->sockaddr.un.sun_path) - 1)
 550                         return -E2BIG;
 551
 552                 b->sockaddr.un.sun_family = AF_UNIX;
 553                 b->sockaddr.un.sun_path[0] = 0;
 554                 strncpy(b->sockaddr.un.sun_path+1, abstract, sizeof(b->sockaddr.un.sun_path)-1);
 555                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + 1 + l;
 556         }
 557
 558         return 0;
 559 }
 560
 561 static int parse_tcp_address(sd_bus *b, const char **p, char **guid) {
 562         _cleanup_free_ char *host = NULL, *port = NULL, *family = NULL;
 563         int r;
 564         struct addrinfo *result, hints = {
 565                 .ai_socktype = SOCK_STREAM,
 566                 .ai_flags = AI_ADDRCONFIG,
 567         };
 568
 569         assert(b);
 570         assert(p);
 571         assert(*p);
 572         assert(guid);
 573
 574         while (**p != 0 && **p != ';') {
 575                 r = parse_address_key(p, "guid", guid);
 576                 if (r < 0)
 577                         return r;
 578                 else if (r > 0)
 579                         continue;
 580
 581                 r = parse_address_key(p, "host", &host);
 582                 if (r < 0)
 583                         return r;
 584                 else if (r > 0)
 585                         continue;
 586
 587                 r = parse_address_key(p, "port", &port);
 588                 if (r < 0)
 589                         return r;
 590                 else if (r > 0)
 591                         continue;
 592
 593                 r = parse_address_key(p, "family", &family);
 594                 if (r < 0)
 595                         return r;
 596                 else if (r > 0)
 597                         continue;
 598
 599                 skip_address_key(p);
 600         }
 601
 602         if (!host || !port)
 603                 return -EINVAL;
 604
 605         if (family) {
 606                 if (streq(family, "ipv4"))
 607                         hints.ai_family = AF_INET;
 608                 else if (streq(family, "ipv6"))
 609                         hints.ai_family = AF_INET6;
 610                 else
 611                         return -EINVAL;
 612         }
 613
 614         r = getaddrinfo(host, port, &hints, &result);
 615         if (r == EAI_SYSTEM)
 616                 return -errno;
 617         else if (r != 0)
 618                 return -EADDRNOTAVAIL;
 619
 620         memcpy(&b->sockaddr, result->ai_addr, result->ai_addrlen);
 621         b->sockaddr_size = result->ai_addrlen;
 622
 623         freeaddrinfo(result);
 624
 625         return 0;
 626 }
 627
 628 static int parse_exec_address(sd_bus *b, const char **p, char **guid) {
 629         char *path = NULL;
 630         unsigned n_argv = 0, j;
 631         char **argv = NULL;
 632         int r;
 633
 634         assert(b);
 635         assert(p);
 636         assert(*p);
 637         assert(guid);
 638
 639         while (**p != 0 && **p != ';') {
 640                 r = parse_address_key(p, "guid", guid);
 641                 if (r < 0)
 642                         goto fail;
 643                 else if (r > 0)
 644                         continue;
 645
 646                 r = parse_address_key(p, "path", &path);
 647                 if (r < 0)
 648                         goto fail;
 649                 else if (r > 0)
 650                         continue;
 651
 652                 if (startswith(*p, "argv")) {
 653                         unsigned ul;
 654
 655                         errno = 0;
 656                         ul = strtoul(*p + 4, (char**) p, 10);
 657                         if (errno > 0 || **p != '=' || ul > 256) {
 658                                 r = -EINVAL;
 659                                 goto fail;
 660                         }
 661
 662                         (*p) ++;
 663
 664                         if (ul >= n_argv) {
 665                                 char **x;
 666
 667                                 x = realloc(argv, sizeof(char*) * (ul + 2));
 668                                 if (!x) {
 669                                         r = -ENOMEM;
 670                                         goto fail;
 671                                 }
 672
 673                                 memset(x + n_argv, 0, sizeof(char*) * (ul - n_argv + 2));
 674
 675                                 argv = x;
 676                                 n_argv = ul + 1;
 677                         }
 678
 679                         r = parse_address_key(p, NULL, argv + ul);
 680                         if (r < 0)
 681                                 goto fail;
 682
 683                         continue;
 684                 }
 685
 686                 skip_address_key(p);
 687         }
 688
 689         if (!path) {
 690                 r = -EINVAL;
 691                 goto fail;
 692         }
 693
 694         /* Make sure there are no holes in the array, with the
 695          * exception of argv[0] */
 696         for (j = 1; j < n_argv; j++)
 697                 if (!argv[j]) {
 698                         r = -EINVAL;
 699                         goto fail;
 700                 }
 701
 702         if (argv && argv[0] == NULL) {
 703                 argv[0] = strdup(path);
 704                 if (!argv[0]) {
 705                         r = -ENOMEM;
 706                         goto fail;
 707                 }
 708         }
 709
 710         b->exec_path = path;
 711         b->exec_argv = argv;
 712         return 0;
 713
 714 fail:
 715         for (j = 0; j < n_argv; j++)
 716                 free(argv[j]);
 717
 718         free(argv);
 719         free(path);
 720         return r;
 721 }
 722
 723 static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
 724         _cleanup_free_ char *path = NULL;
 725         int r;
 726
 727         assert(b);
 728         assert(p);
 729         assert(*p);
 730         assert(guid);
 731
 732         while (**p != 0 && **p != ';') {
 733                 r = parse_address_key(p, "guid", guid);
 734                 if (r < 0)
 735                         return r;
 736                 else if (r > 0)
 737                         continue;
 738
 739                 r = parse_address_key(p, "path", &path);
 740                 if (r < 0)
 741                         return r;
 742                 else if (r > 0)
 743                         continue;
 744
 745                 skip_address_key(p);
 746         }
 747
 748         if (!path)
 749                 return -EINVAL;
 750
 751         free(b->kernel);
 752         b->kernel = path;
 753         path = NULL;
 754
 755         return 0;
 756 }
 757
 758 static int parse_container_address(sd_bus *b, const char **p, char **guid) {
 759         _cleanup_free_ char *machine = NULL;
 760         int r;
 761
 762         assert(b);
 763         assert(p);
 764         assert(*p);
 765         assert(guid);
 766
 767         while (**p != 0 && **p != ';') {
 768                 r = parse_address_key(p, "guid", guid);
 769                 if (r < 0)
 770                         return r;
 771                 else if (r > 0)
 772                         continue;
 773
 774                 r = parse_address_key(p, "machine", &machine);
 775                 if (r < 0)
 776                         return r;
 777                 else if (r > 0)
 778                         continue;
 779
 780                 skip_address_key(p);
 781         }
 782
 783         if (!machine)
 784                 return -EINVAL;
 785
 786         free(b->machine);
 787         b->machine = machine;
 788         machine = NULL;
 789
 790         b->sockaddr.un.sun_family = AF_UNIX;
 791         strncpy(b->sockaddr.un.sun_path, "/var/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
 792         b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/var/run/dbus/system_bus_socket") - 1;
 793
 794         return 0;
 795 }
 796
 797 static void bus_reset_parsed_address(sd_bus *b) {
 798         assert(b);
 799
 800         zero(b->sockaddr);
 801         b->sockaddr_size = 0;
 802         strv_free(b->exec_argv);
 803         free(b->exec_path);
 804         b->exec_path = NULL;
 805         b->exec_argv = NULL;
 806         b->server_id = SD_ID128_NULL;
 807         free(b->kernel);
 808         b->kernel = NULL;
 809         free(b->machine);
 810         b->machine = NULL;
 811 }
 812
 813 static int bus_parse_next_address(sd_bus *b) {
 814         _cleanup_free_ char *guid = NULL;
 815         const char *a;
 816         int r;
 817
 818         assert(b);
 819
 820         if (!b->address)
 821                 return 0;
 822         if (b->address[b->address_index] == 0)
 823                 return 0;
 824
 825         bus_reset_parsed_address(b);
 826
 827         a = b->address + b->address_index;
 828
 829         while (*a != 0) {
 830
 831                 if (*a == ';') {
 832                         a++;
 833                         continue;
 834                 }
 835
 836                 if (startswith(a, "unix:")) {
 837                         a += 5;
 838
 839                         r = parse_unix_address(b, &a, &guid);
 840                         if (r < 0)
 841                                 return r;
 842                         break;
 843
 844                 } else if (startswith(a, "tcp:")) {
 845
 846                         a += 4;
 847                         r = parse_tcp_address(b, &a, &guid);
 848                         if (r < 0)
 849                                 return r;
 850
 851                         break;
 852
 853                 } else if (startswith(a, "unixexec:")) {
 854
 855                         a += 9;
 856                         r = parse_exec_address(b, &a, &guid);
 857                         if (r < 0)
 858                                 return r;
 859
 860                         break;
 861
 862                 } else if (startswith(a, "kernel:")) {
 863
 864                         a += 7;
 865                         r = parse_kernel_address(b, &a, &guid);
 866                         if (r < 0)
 867                                 return r;
 868
 869                         break;
 870                 } else if (startswith(a, "x-container:")) {
 871
 872                         a += 12;
 873                         r = parse_container_address(b, &a, &guid);
 874                         if (r < 0)
 875                                 return r;
 876
 877                         break;
 878                 }
 879
 880                 a = strchr(a, ';');
 881                 if (!a)
 882                         return 0;
 883         }
 884
 885         if (guid) {
 886                 r = sd_id128_from_string(guid, &b->server_id);
 887                 if (r < 0)
 888                         return r;
 889         }
 890
 891         b->address_index = a - b->address;
 892         return 1;
 893 }
 894
 895 static int bus_start_address(sd_bus *b) {
 896         int r;
 897
 898         assert(b);
 899
 900         for (;;) {
 901                 sd_bus_close(b);
 902
 903                 if (b->exec_path) {
 904
 905                         r = bus_socket_exec(b);
 906                         if (r >= 0)
 907                                 return r;
 908
 909                         b->last_connect_error = -r;
 910                 } else if (b->kernel) {
 911
 912                         r = bus_kernel_connect(b);
 913                         if (r >= 0)
 914                                 return r;
 915
 916                         b->last_connect_error = -r;
 917
 918                 } else if (b->machine) {
 919
 920                         r = bus_container_connect(b);
 921                         if (r >= 0)
 922                                 return r;
 923
 924                         b->last_connect_error = -r;
 925
 926                 } else if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
 927
 928                         r = bus_socket_connect(b);
 929                         if (r >= 0)
 930                                 return r;
 931
 932                         b->last_connect_error = -r;
 933                 }
 934
 935                 r = bus_parse_next_address(b);
 936                 if (r < 0)
 937                         return r;
 938                 if (r == 0)
 939                         return b->last_connect_error ? -b->last_connect_error : -ECONNREFUSED;
 940         }
 941 }
 942
 943 int bus_next_address(sd_bus *b) {
 944         assert(b);
 945
 946         bus_reset_parsed_address(b);
 947         return bus_start_address(b);
 948 }
 949
 950 static int bus_start_fd(sd_bus *b) {
 951         struct stat st;
 952         int r;
 953
 954         assert(b);
 955         assert(b->input_fd >= 0);
 956         assert(b->output_fd >= 0);
 957
 958         r = fd_nonblock(b->input_fd, true);
 959         if (r < 0)
 960                 return r;
 961
 962         r = fd_cloexec(b->input_fd, true);
 963         if (r < 0)
 964                 return r;
 965
 966         if (b->input_fd != b->output_fd) {
 967                 r = fd_nonblock(b->output_fd, true);
 968                 if (r < 0)
 969                         return r;
 970
 971                 r = fd_cloexec(b->output_fd, true);
 972                 if (r < 0)
 973                         return r;
 974         }
 975
 976         if (fstat(b->input_fd, &st) < 0)
 977                 return -errno;
 978
 979         if (S_ISCHR(b->input_fd))
 980                 return bus_kernel_take_fd(b);
 981         else
 982                 return bus_socket_take_fd(b);
 983 }
 984
 985 int sd_bus_start(sd_bus *bus) {
 986         int r;
 987
 988         assert_return(bus, -EINVAL);
 989         assert_return(bus->state == BUS_UNSET, -EPERM);
 990         assert_return(!bus_pid_changed(bus), -ECHILD);
 991
 992         bus->state = BUS_OPENING;
 993
 994         if (bus->is_server && bus->bus_client)
 995                 return -EINVAL;
 996
 997         if (bus->input_fd >= 0)
 998                 r = bus_start_fd(bus);
 999         else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel || bus->machine)
1000                 r = bus_start_address(bus);
1001         else
1002                 return -EINVAL;
1003
1004         if (r < 0)
1005                 return r;
1006
1007         return bus_send_hello(bus);
1008 }
1009
1010 int sd_bus_open_system(sd_bus **ret) {
1011         const char *e;
1012         sd_bus *b;
1013         int r;
1014
1015         assert_return(ret, -EINVAL);
1016
1017         r = sd_bus_new(&b);
1018         if (r < 0)
1019                 return r;
1020
1021         e = secure_getenv("DBUS_SYSTEM_BUS_ADDRESS");
1022         if (e) {
1023                 r = sd_bus_set_address(b, e);
1024                 if (r < 0)
1025                         goto fail;
1026         } else {
1027                 b->sockaddr.un.sun_family = AF_UNIX;
1028                 strncpy(b->sockaddr.un.sun_path, "/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
1029                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/run/dbus/system_bus_socket") - 1;
1030         }
1031
1032         b->bus_client = true;
1033
1034         r = sd_bus_start(b);
1035         if (r < 0)
1036                 goto fail;
1037
1038         *ret = b;
1039         return 0;
1040
1041 fail:
1042         bus_free(b);
1043         return r;
1044 }
1045
1046 int sd_bus_open_user(sd_bus **ret) {
1047         const char *e;
1048         sd_bus *b;
1049         size_t l;
1050         int r;
1051
1052         assert_return(ret, -EINVAL);
1053
1054         r = sd_bus_new(&b);
1055         if (r < 0)
1056                 return r;
1057
1058         e = secure_getenv("DBUS_SESSION_BUS_ADDRESS");
1059         if (e) {
1060                 r = sd_bus_set_address(b, e);
1061                 if (r < 0)
1062                         goto fail;
1063         } else {
1064                 e = secure_getenv("XDG_RUNTIME_DIR");
1065                 if (!e) {
1066                         r = -ENOENT;
1067                         goto fail;
1068                 }
1069
1070                 l = strlen(e);
1071                 if (l + 4 > sizeof(b->sockaddr.un.sun_path)) {
1072                         r = -E2BIG;
1073                         goto fail;
1074                 }
1075
1076                 b->sockaddr.un.sun_family = AF_UNIX;
1077                 memcpy(mempcpy(b->sockaddr.un.sun_path, e, l), "/bus", 4);
1078                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l + 4;
1079         }
1080
1081         b->bus_client = true;
1082
1083         r = sd_bus_start(b);
1084         if (r < 0)
1085                 goto fail;
1086
1087         *ret = b;
1088         return 0;
1089
1090 fail:
1091         bus_free(b);
1092         return r;
1093 }
1094
1095 int sd_bus_open_system_remote(const char *host, sd_bus **ret) {
1096         _cleanup_free_ char *e = NULL;
1097         char *p = NULL;
1098         sd_bus *bus;
1099         int r;
1100
1101         assert_return(host, -EINVAL);
1102         assert_return(ret, -EINVAL);
1103
1104         e = bus_address_escape(host);
1105         if (!e)
1106                 return -ENOMEM;
1107
1108         p = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", NULL);
1109         if (!p)
1110                 return -ENOMEM;
1111
1112         r = sd_bus_new(&bus);
1113         if (r < 0) {
1114                 free(p);
1115                 return r;
1116         }
1117
1118         bus->address = p;
1119         bus->bus_client = true;
1120
1121         r = sd_bus_start(bus);
1122         if (r < 0) {
1123                 bus_free(bus);
1124                 return r;
1125         }
1126
1127         *ret = bus;
1128         return 0;
1129 }
1130
1131 int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
1132         _cleanup_free_ char *e = NULL;
1133         sd_bus *bus;
1134         char *p;
1135         int r;
1136
1137         assert_return(machine, -EINVAL);
1138         assert_return(ret, -EINVAL);
1139
1140         e = bus_address_escape(machine);
1141         if (!e)
1142                 return -ENOMEM;
1143
1144         p = strjoin("x-container:machine=", e, NULL);
1145         if (!p)
1146                 return -ENOMEM;
1147
1148         r = sd_bus_new(&bus);
1149         if (r < 0) {
1150                 free(p);
1151                 return r;
1152         }
1153
1154         bus->address = p;
1155         bus->bus_client = true;
1156
1157         r = sd_bus_start(bus);
1158         if (r < 0) {
1159                 bus_free(bus);
1160                 return r;
1161         }
1162
1163         *ret = bus;
1164         return 0;
1165 }
1166
1167 void sd_bus_close(sd_bus *bus) {
1168         if (!bus)
1169                 return;
1170         if (bus->state == BUS_CLOSED)
1171                 return;
1172         if (bus_pid_changed(bus))
1173                 return;
1174
1175         bus->state = BUS_CLOSED;
1176
1177         sd_bus_detach_event(bus);
1178
1179         if (!bus->is_kernel)
1180                 bus_close_fds(bus);
1181
1182         /* We'll leave the fd open in case this is a kernel bus, since
1183          * there might still be memblocks around that reference this
1184          * bus, and they might need to invoke the
1185          * KDBUS_CMD_MSG_RELEASE ioctl on the fd when they are
1186          * freed. */
1187 }
1188
1189 sd_bus *sd_bus_ref(sd_bus *bus) {
1190         if (!bus)
1191                 return NULL;
1192
1193         assert_se(REFCNT_INC(bus->n_ref) >= 2);
1194
1195         return bus;
1196 }
1197
1198 sd_bus *sd_bus_unref(sd_bus *bus) {
1199         if (!bus)
1200                 return NULL;
1201
1202         if (REFCNT_DEC(bus->n_ref) <= 0)
1203                 bus_free(bus);
1204
1205         return NULL;
1206 }
1207
1208 int sd_bus_is_open(sd_bus *bus) {
1209
1210         assert_return(bus, -EINVAL);
1211         assert_return(!bus_pid_changed(bus), -ECHILD);
1212
1213         return BUS_IS_OPEN(bus->state);
1214 }
1215
1216 int sd_bus_can_send(sd_bus *bus, char type) {
1217         int r;
1218
1219         assert_return(bus, -EINVAL);
1220         assert_return(bus->state != BUS_UNSET, -ENOTCONN);
1221         assert_return(!bus_pid_changed(bus), -ECHILD);
1222
1223         if (type == SD_BUS_TYPE_UNIX_FD) {
1224                 if (!(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD))
1225                         return 0;
1226
1227                 r = bus_ensure_running(bus);
1228                 if (r < 0)
1229                         return r;
1230
1231                 return bus->can_fds;
1232         }
1233
1234         return bus_type_is_valid(type);
1235 }
1236
1237 int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
1238         int r;
1239
1240         assert_return(bus, -EINVAL);
1241         assert_return(server_id, -EINVAL);
1242         assert_return(!bus_pid_changed(bus), -ECHILD);
1243
1244         r = bus_ensure_running(bus);
1245         if (r < 0)
1246                 return r;
1247
1248         *server_id = bus->server_id;
1249         return 0;
1250 }
1251
1252 static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
1253         assert(m);
1254
1255         if (m->header->version > b->message_version)
1256                 return -EPERM;
1257
1258         if (m->sealed)
1259                 return 0;
1260
1261         return bus_message_seal(m, ++b->serial);
1262 }
1263
1264 static int dispatch_wqueue(sd_bus *bus) {
1265         int r, ret = 0;
1266
1267         assert(bus);
1268         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1269
1270         while (bus->wqueue_size > 0) {
1271
1272                 if (bus->is_kernel)
1273                         r = bus_kernel_write_message(bus, bus->wqueue[0]);
1274                 else
1275                         r = bus_socket_write_message(bus, bus->wqueue[0], &bus->windex);
1276
1277                 if (r < 0) {
1278                         sd_bus_close(bus);
1279                         return r;
1280                 } else if (r == 0)
1281                         /* Didn't do anything this time */
1282                         return ret;
1283                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1284                         /* Fully written. Let's drop the entry from
1285                          * the queue.
1286                          *
1287                          * This isn't particularly optimized, but
1288                          * well, this is supposed to be our worst-case
1289                          * buffer only, and the socket buffer is
1290                          * supposed to be our primary buffer, and if
1291                          * it got full, then all bets are off
1292                          * anyway. */
1293
1294                         sd_bus_message_unref(bus->wqueue[0]);
1295                         bus->wqueue_size --;
1296                         memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1297                         bus->windex = 0;
1298
1299                         ret = 1;
1300                 }
1301         }
1302
1303         return ret;
1304 }
1305
1306 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
1307         sd_bus_message *z = NULL;
1308         int r, ret = 0;
1309
1310         assert(bus);
1311         assert(m);
1312         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1313
1314         if (bus->rqueue_size > 0) {
1315                 /* Dispatch a queued message */
1316
1317                 *m = bus->rqueue[0];
1318                 bus->rqueue_size --;
1319                 memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1320                 return 1;
1321         }
1322
1323         /* Try to read a new message */
1324         do {
1325                 if (bus->is_kernel)
1326                         r = bus_kernel_read_message(bus, &z);
1327                 else
1328                         r = bus_socket_read_message(bus, &z);
1329
1330                 if (r < 0) {
1331                         sd_bus_close(bus);
1332                         return r;
1333                 }
1334                 if (r == 0)
1335                         return ret;
1336
1337                 ret = 1;
1338         } while (!z);
1339
1340         *m = z;
1341         return ret;
1342 }
1343
1344 int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
1345         int r;
1346
1347         assert_return(bus, -EINVAL);
1348         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1349         assert_return(m, -EINVAL);
1350         assert_return(!bus_pid_changed(bus), -ECHILD);
1351
1352         if (m->n_fds > 0) {
1353                 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1354                 if (r < 0)
1355                         return r;
1356                 if (r == 0)
1357                         return -ENOTSUP;
1358         }
1359
1360         /* If the serial number isn't kept, then we know that no reply
1361          * is expected */
1362         if (!serial && !m->sealed)
1363                 m->header->flags |= SD_BUS_MESSAGE_NO_REPLY_EXPECTED;
1364
1365         r = bus_seal_message(bus, m);
1366         if (r < 0)
1367                 return r;
1368
1369         /* If this is a reply and no reply was requested, then let's
1370          * suppress this, if we can */
1371         if (m->dont_send && !serial)
1372                 return 1;
1373
1374         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1375                 size_t idx = 0;
1376
1377                 if (bus->is_kernel)
1378                         r = bus_kernel_write_message(bus, m);
1379                 else
1380                         r = bus_socket_write_message(bus, m, &idx);
1381
1382                 if (r < 0) {
1383                         sd_bus_close(bus);
1384                         return r;
1385                 } else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
1386                         /* Wasn't fully written. So let's remember how
1387                          * much was written. Note that the first entry
1388                          * of the wqueue array is always allocated so
1389                          * that we always can remember how much was
1390                          * written. */
1391                         bus->wqueue[0] = sd_bus_message_ref(m);
1392                         bus->wqueue_size = 1;
1393                         bus->windex = idx;
1394                 }
1395         } else {
1396                 sd_bus_message **q;
1397
1398                 /* Just append it to the queue. */
1399
1400                 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1401                         return -ENOBUFS;
1402
1403                 q = realloc(bus->wqueue, sizeof(sd_bus_message*) * (bus->wqueue_size + 1));
1404                 if (!q)
1405                         return -ENOMEM;
1406
1407                 bus->wqueue = q;
1408                 q[bus->wqueue_size ++] = sd_bus_message_ref(m);
1409         }
1410
1411         if (serial)
1412                 *serial = BUS_MESSAGE_SERIAL(m);
1413
1414         return 1;
1415 }
1416
1417 static usec_t calc_elapse(uint64_t usec) {
1418         if (usec == (uint64_t) -1)
1419                 return 0;
1420
1421         if (usec == 0)
1422                 usec = BUS_DEFAULT_TIMEOUT;
1423
1424         return now(CLOCK_MONOTONIC) + usec;
1425 }
1426
1427 static int timeout_compare(const void *a, const void *b) {
1428         const struct reply_callback *x = a, *y = b;
1429
1430         if (x->timeout != 0 && y->timeout == 0)
1431                 return -1;
1432
1433         if (x->timeout == 0 && y->timeout != 0)
1434                 return 1;
1435
1436         if (x->timeout < y->timeout)
1437                 return -1;
1438
1439         if (x->timeout > y->timeout)
1440                 return 1;
1441
1442         return 0;
1443 }
1444
1445 int sd_bus_send_with_reply(
1446                 sd_bus *bus,
1447                 sd_bus_message *m,
1448                 sd_bus_message_handler_t callback,
1449                 void *userdata,
1450                 uint64_t usec,
1451                 uint64_t *serial) {
1452
1453         struct reply_callback *c;
1454         int r;
1455
1456         assert_return(bus, -EINVAL);
1457         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1458         assert_return(m, -EINVAL);
1459         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1460         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1461         assert_return(callback, -EINVAL);
1462         assert_return(!bus_pid_changed(bus), -ECHILD);
1463
1464         r = hashmap_ensure_allocated(&bus->reply_callbacks, uint64_hash_func, uint64_compare_func);
1465         if (r < 0)
1466                 return r;
1467
1468         if (usec != (uint64_t) -1) {
1469                 r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1470                 if (r < 0)
1471                         return r;
1472         }
1473
1474         r = bus_seal_message(bus, m);
1475         if (r < 0)
1476                 return r;
1477
1478         c = new0(struct reply_callback, 1);
1479         if (!c)
1480                 return -ENOMEM;
1481
1482         c->callback = callback;
1483         c->userdata = userdata;
1484         c->serial = BUS_MESSAGE_SERIAL(m);
1485         c->timeout = calc_elapse(usec);
1486
1487         r = hashmap_put(bus->reply_callbacks, &c->serial, c);
1488         if (r < 0) {
1489                 free(c);
1490                 return r;
1491         }
1492
1493         if (c->timeout != 0) {
1494                 r = prioq_put(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1495                 if (r < 0) {
1496                         c->timeout = 0;
1497                         sd_bus_send_with_reply_cancel(bus, c->serial);
1498                         return r;
1499                 }
1500         }
1501
1502         r = sd_bus_send(bus, m, serial);
1503         if (r < 0) {
1504                 sd_bus_send_with_reply_cancel(bus, c->serial);
1505                 return r;
1506         }
1507
1508         return r;
1509 }
1510
1511 int sd_bus_send_with_reply_cancel(sd_bus *bus, uint64_t serial) {
1512         struct reply_callback *c;
1513
1514         assert_return(bus, -EINVAL);
1515         assert_return(serial != 0, -EINVAL);
1516         assert_return(!bus_pid_changed(bus), -ECHILD);
1517
1518         c = hashmap_remove(bus->reply_callbacks, &serial);
1519         if (!c)
1520                 return 0;
1521
1522         if (c->timeout != 0)
1523                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1524
1525         free(c);
1526         return 1;
1527 }
1528
1529 int bus_ensure_running(sd_bus *bus) {
1530         int r;
1531
1532         assert(bus);
1533
1534         if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED)
1535                 return -ENOTCONN;
1536         if (bus->state == BUS_RUNNING)
1537                 return 1;
1538
1539         for (;;) {
1540                 r = sd_bus_process(bus, NULL);
1541                 if (r < 0)
1542                         return r;
1543                 if (bus->state == BUS_RUNNING)
1544                         return 1;
1545                 if (r > 0)
1546                         continue;
1547
1548                 r = sd_bus_wait(bus, (uint64_t) -1);
1549                 if (r < 0)
1550                         return r;
1551         }
1552 }
1553
1554 int sd_bus_send_with_reply_and_block(
1555                 sd_bus *bus,
1556                 sd_bus_message *m,
1557                 uint64_t usec,
1558                 sd_bus_error *error,
1559                 sd_bus_message **reply) {
1560
1561         int r;
1562         usec_t timeout;
1563         uint64_t serial;
1564         bool room = false;
1565
1566         assert_return(bus, -EINVAL);
1567         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1568         assert_return(m, -EINVAL);
1569         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1570         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1571         assert_return(!bus_error_is_dirty(error), -EINVAL);
1572         assert_return(!bus_pid_changed(bus), -ECHILD);
1573
1574         r = bus_ensure_running(bus);
1575         if (r < 0)
1576                 return r;
1577
1578         r = sd_bus_send(bus, m, &serial);
1579         if (r < 0)
1580                 return r;
1581
1582         timeout = calc_elapse(usec);
1583
1584         for (;;) {
1585                 usec_t left;
1586                 sd_bus_message *incoming = NULL;
1587
1588                 if (!room) {
1589                         sd_bus_message **q;
1590
1591                         if (bus->rqueue_size >= BUS_RQUEUE_MAX)
1592                                 return -ENOBUFS;
1593
1594                         /* Make sure there's room for queuing this
1595                          * locally, before we read the message */
1596
1597                         q = realloc(bus->rqueue, (bus->rqueue_size + 1) * sizeof(sd_bus_message*));
1598                         if (!q)
1599                                 return -ENOMEM;
1600
1601                         bus->rqueue = q;
1602                         room = true;
1603                 }
1604
1605                 if (bus->is_kernel)
1606                         r = bus_kernel_read_message(bus, &incoming);
1607                 else
1608                         r = bus_socket_read_message(bus, &incoming);
1609                 if (r < 0)
1610                         return r;
1611                 if (incoming) {
1612
1613                         if (incoming->reply_serial == serial) {
1614                                 /* Found a match! */
1615
1616                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
1617
1618                                         if (reply)
1619                                                 *reply = incoming;
1620                                         else
1621                                                 sd_bus_message_unref(incoming);
1622
1623                                         return 1;
1624                                 }
1625
1626                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR) {
1627                                         int k;
1628
1629                                         r = sd_bus_error_copy(error, &incoming->error);
1630                                         if (r < 0) {
1631                                                 sd_bus_message_unref(incoming);
1632                                                 return r;
1633                                         }
1634
1635                                         k = sd_bus_error_get_errno(&incoming->error);
1636                                         sd_bus_message_unref(incoming);
1637                                         return -k;
1638                                 }
1639
1640                                 sd_bus_message_unref(incoming);
1641                                 return -EIO;
1642                         }
1643
1644                         /* There's already guaranteed to be room for
1645                          * this, so need to resize things here */
1646                         bus->rqueue[bus->rqueue_size ++] = incoming;
1647                         room = false;
1648
1649                         /* Try to read more, right-away */
1650                         continue;
1651                 }
1652                 if (r != 0)
1653                         continue;
1654
1655                 if (timeout > 0) {
1656                         usec_t n;
1657
1658                         n = now(CLOCK_MONOTONIC);
1659                         if (n >= timeout)
1660                                 return -ETIMEDOUT;
1661
1662                         left = timeout - n;
1663                 } else
1664                         left = (uint64_t) -1;
1665
1666                 r = bus_poll(bus, true, left);
1667                 if (r < 0)
1668                         return r;
1669
1670                 r = dispatch_wqueue(bus);
1671                 if (r < 0)
1672                         return r;
1673         }
1674 }
1675
1676 int sd_bus_get_fd(sd_bus *bus) {
1677
1678         assert_return(bus, -EINVAL);
1679         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1680         assert_return(bus->input_fd == bus->output_fd, -EPERM);
1681         assert_return(!bus_pid_changed(bus), -ECHILD);
1682
1683         return bus->input_fd;
1684 }
1685
1686 int sd_bus_get_events(sd_bus *bus) {
1687         int flags = 0;
1688
1689         assert_return(bus, -EINVAL);
1690         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1691         assert_return(!bus_pid_changed(bus), -ECHILD);
1692
1693         if (bus->state == BUS_OPENING)
1694                 flags |= POLLOUT;
1695         else if (bus->state == BUS_AUTHENTICATING) {
1696
1697                 if (bus_socket_auth_needs_write(bus))
1698                         flags |= POLLOUT;
1699
1700                 flags |= POLLIN;
1701
1702         } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
1703                 if (bus->rqueue_size <= 0)
1704                         flags |= POLLIN;
1705                 if (bus->wqueue_size > 0)
1706                         flags |= POLLOUT;
1707         }
1708
1709         return flags;
1710 }
1711
1712 int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
1713         struct reply_callback *c;
1714
1715         assert_return(bus, -EINVAL);
1716         assert_return(timeout_usec, -EINVAL);
1717         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1718         assert_return(!bus_pid_changed(bus), -ECHILD);
1719
1720         if (bus->state == BUS_AUTHENTICATING) {
1721                 *timeout_usec = bus->auth_timeout;
1722                 return 1;
1723         }
1724
1725         if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
1726                 *timeout_usec = (uint64_t) -1;
1727                 return 0;
1728         }
1729
1730         if (bus->rqueue_size > 0) {
1731                 *timeout_usec = 0;
1732                 return 1;
1733         }
1734
1735         c = prioq_peek(bus->reply_callbacks_prioq);
1736         if (!c) {
1737                 *timeout_usec = (uint64_t) -1;
1738                 return 0;
1739         }
1740
1741         *timeout_usec = c->timeout;
1742         return 1;
1743 }
1744
1745 static int process_timeout(sd_bus *bus) {
1746         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
1747         struct reply_callback *c;
1748         usec_t n;
1749         int r;
1750
1751         assert(bus);
1752
1753         c = prioq_peek(bus->reply_callbacks_prioq);
1754         if (!c)
1755                 return 0;
1756
1757         n = now(CLOCK_MONOTONIC);
1758         if (c->timeout > n)
1759                 return 0;
1760
1761         r = bus_message_new_synthetic_error(
1762                         bus,
1763                         c->serial,
1764                         &SD_BUS_ERROR_MAKE(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
1765                         &m);
1766         if (r < 0)
1767                 return r;
1768
1769         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
1770         hashmap_remove(bus->reply_callbacks, &c->serial);
1771
1772         r = c->callback(bus, m, c->userdata);
1773         free(c);
1774
1775         return r < 0 ? r : 1;
1776 }
1777
1778 static int process_hello(sd_bus *bus, sd_bus_message *m) {
1779         assert(bus);
1780         assert(m);
1781
1782         if (bus->state != BUS_HELLO)
1783                 return 0;
1784
1785         /* Let's make sure the first message on the bus is the HELLO
1786          * reply. But note that we don't actually parse the message
1787          * here (we leave that to the usual handling), we just verify
1788          * we don't let any earlier msg through. */
1789
1790         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1791             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1792                 return -EIO;
1793
1794         if (m->reply_serial != bus->hello_serial)
1795                 return -EIO;
1796
1797         return 0;
1798 }
1799
1800 static int process_reply(sd_bus *bus, sd_bus_message *m) {
1801         struct reply_callback *c;
1802         int r;
1803
1804         assert(bus);
1805         assert(m);
1806
1807         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1808             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1809                 return 0;
1810
1811         c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
1812         if (!c)
1813                 return 0;
1814
1815         if (c->timeout != 0)
1816                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1817
1818         r = sd_bus_message_rewind(m, true);
1819         if (r < 0)
1820                 return r;
1821
1822         r = c->callback(bus, m, c->userdata);
1823         free(c);
1824
1825         return r;
1826 }
1827
1828 static int process_filter(sd_bus *bus, sd_bus_message *m) {
1829         struct filter_callback *l;
1830         int r;
1831
1832         assert(bus);
1833         assert(m);
1834
1835         do {
1836                 bus->filter_callbacks_modified = false;
1837
1838                 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
1839
1840                         if (bus->filter_callbacks_modified)
1841                                 break;
1842
1843                         /* Don't run this more than once per iteration */
1844                         if (l->last_iteration == bus->iteration_counter)
1845                                 continue;
1846
1847                         l->last_iteration = bus->iteration_counter;
1848
1849                         r = sd_bus_message_rewind(m, true);
1850                         if (r < 0)
1851                                 return r;
1852
1853                         r = l->callback(bus, m, l->userdata);
1854                         if (r != 0)
1855                                 return r;
1856
1857                 }
1858
1859         } while (bus->filter_callbacks_modified);
1860
1861         return 0;
1862 }
1863
1864 static int process_match(sd_bus *bus, sd_bus_message *m) {
1865         int r;
1866
1867         assert(bus);
1868         assert(m);
1869
1870         do {
1871                 bus->match_callbacks_modified = false;
1872
1873                 r = bus_match_run(bus, &bus->match_callbacks, m);
1874                 if (r != 0)
1875                         return r;
1876
1877         } while (bus->match_callbacks_modified);
1878
1879         return 0;
1880 }
1881
1882 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
1883         _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
1884         int r;
1885
1886         assert(bus);
1887         assert(m);
1888
1889         if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
1890                 return 0;
1891
1892         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
1893                 return 0;
1894
1895         if (m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED)
1896                 return 1;
1897
1898         if (streq_ptr(m->member, "Ping"))
1899                 r = sd_bus_message_new_method_return(bus, m, &reply);
1900         else if (streq_ptr(m->member, "GetMachineId")) {
1901                 sd_id128_t id;
1902                 char sid[33];
1903
1904                 r = sd_id128_get_machine(&id);
1905                 if (r < 0)
1906                         return r;
1907
1908                 r = sd_bus_message_new_method_return(bus, m, &reply);
1909                 if (r < 0)
1910                         return r;
1911
1912                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
1913         } else {
1914                 r = sd_bus_message_new_method_errorf(
1915                                 bus, m, &reply,
1916                                 SD_BUS_ERROR_UNKNOWN_METHOD,
1917                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
1918         }
1919
1920         if (r < 0)
1921                 return r;
1922
1923         r = sd_bus_send(bus, reply, NULL);
1924         if (r < 0)
1925                 return r;
1926
1927         return 1;
1928 }
1929
1930 static int process_message(sd_bus *bus, sd_bus_message *m) {
1931         int r;
1932
1933         assert(bus);
1934         assert(m);
1935
1936         bus->iteration_counter++;
1937
1938         log_debug("Got message sender=%s object=%s interface=%s member=%s",
1939                   strna(sd_bus_message_get_sender(m)),
1940                   strna(sd_bus_message_get_path(m)),
1941                   strna(sd_bus_message_get_interface(m)),
1942                   strna(sd_bus_message_get_member(m)));
1943
1944         r = process_hello(bus, m);
1945         if (r != 0)
1946                 return r;
1947
1948         r = process_reply(bus, m);
1949         if (r != 0)
1950                 return r;
1951
1952         r = process_filter(bus, m);
1953         if (r != 0)
1954                 return r;
1955
1956         r = process_match(bus, m);
1957         if (r != 0)
1958                 return r;
1959
1960         r = process_builtin(bus, m);
1961         if (r != 0)
1962                 return r;
1963
1964         return bus_process_object(bus, m);
1965 }
1966
1967 static int process_running(sd_bus *bus, sd_bus_message **ret) {
1968         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
1969         int r;
1970
1971         assert(bus);
1972         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1973
1974         r = process_timeout(bus);
1975         if (r != 0)
1976                 goto null_message;
1977
1978         r = dispatch_wqueue(bus);
1979         if (r != 0)
1980                 goto null_message;
1981
1982         r = dispatch_rqueue(bus, &m);
1983         if (r < 0)
1984                 return r;
1985         if (!m)
1986                 goto null_message;
1987
1988         r = process_message(bus, m);
1989         if (r != 0)
1990                 goto null_message;
1991
1992         if (ret) {
1993                 r = sd_bus_message_rewind(m, true);
1994                 if (r < 0)
1995                         return r;
1996
1997                 *ret = m;
1998                 m = NULL;
1999                 return 1;
2000         }
2001
2002         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
2003
2004                 r = sd_bus_reply_method_errorf(
2005                                 bus, m,
2006                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
2007                                 "Unknown object '%s'.", m->path);
2008                 if (r < 0)
2009                         return r;
2010         }
2011
2012         return 1;
2013
2014 null_message:
2015         if (r >= 0 && ret)
2016                 *ret = NULL;
2017
2018         return r;
2019 }
2020
2021 int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
2022         BUS_DONT_DESTROY(bus);
2023         int r;
2024
2025         /* Returns 0 when we didn't do anything. This should cause the
2026          * caller to invoke sd_bus_wait() before returning the next
2027          * time. Returns > 0 when we did something, which possibly
2028          * means *ret is filled in with an unprocessed message. */
2029
2030         assert_return(bus, -EINVAL);
2031         assert_return(!bus_pid_changed(bus), -ECHILD);
2032
2033         /* We don't allow recursively invoking sd_bus_process(). */
2034         assert_return(!bus->processing, -EBUSY);
2035
2036         switch (bus->state) {
2037
2038         case BUS_UNSET:
2039         case BUS_CLOSED:
2040                 return -ENOTCONN;
2041
2042         case BUS_OPENING:
2043                 r = bus_socket_process_opening(bus);
2044                 if (r < 0)
2045                         return r;
2046                 if (ret)
2047                         *ret = NULL;
2048                 return r;
2049
2050         case BUS_AUTHENTICATING:
2051
2052                 r = bus_socket_process_authenticating(bus);
2053                 if (r < 0)
2054                         return r;
2055                 if (ret)
2056                         *ret = NULL;
2057                 return r;
2058
2059         case BUS_RUNNING:
2060         case BUS_HELLO:
2061
2062                 bus->processing = true;
2063                 r = process_running(bus, ret);
2064                 bus->processing = false;
2065
2066                 return r;
2067         }
2068
2069         assert_not_reached("Unknown state");
2070 }
2071
2072 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
2073         struct pollfd p[2] = {};
2074         int r, e, n;
2075         struct timespec ts;
2076         usec_t m = (usec_t) -1;
2077
2078         assert(bus);
2079         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2080
2081         e = sd_bus_get_events(bus);
2082         if (e < 0)
2083                 return e;
2084
2085         if (need_more)
2086                 /* The caller really needs some more data, he doesn't
2087                  * care about what's already read, or any timeouts
2088                  * except its own.*/
2089                 e |= POLLIN;
2090         else {
2091                 usec_t until;
2092                 /* The caller wants to process if there's something to
2093                  * process, but doesn't care otherwise */
2094
2095                 r = sd_bus_get_timeout(bus, &until);
2096                 if (r < 0)
2097                         return r;
2098                 if (r > 0) {
2099                         usec_t nw;
2100                         nw = now(CLOCK_MONOTONIC);
2101                         m = until > nw ? until - nw : 0;
2102                 }
2103         }
2104
2105         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
2106                 m = timeout_usec;
2107
2108         p[0].fd = bus->input_fd;
2109         if (bus->output_fd == bus->input_fd) {
2110                 p[0].events = e;
2111                 n = 1;
2112         } else {
2113                 p[0].events = e & POLLIN;
2114                 p[1].fd = bus->output_fd;
2115                 p[1].events = e & POLLOUT;
2116                 n = 2;
2117         }
2118
2119         r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
2120         if (r < 0)
2121                 return -errno;
2122
2123         return r > 0 ? 1 : 0;
2124 }
2125
2126 int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
2127
2128         assert_return(bus, -EINVAL);
2129         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2130         assert_return(!bus_pid_changed(bus), -ECHILD);
2131
2132         if (bus->rqueue_size > 0)
2133                 return 0;
2134
2135         return bus_poll(bus, false, timeout_usec);
2136 }
2137
2138 int sd_bus_flush(sd_bus *bus) {
2139         int r;
2140
2141         assert_return(bus, -EINVAL);
2142         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2143         assert_return(!bus_pid_changed(bus), -ECHILD);
2144
2145         r = bus_ensure_running(bus);
2146         if (r < 0)
2147                 return r;
2148
2149         if (bus->wqueue_size <= 0)
2150                 return 0;
2151
2152         for (;;) {
2153                 r = dispatch_wqueue(bus);
2154                 if (r < 0)
2155                         return r;
2156
2157                 if (bus->wqueue_size <= 0)
2158                         return 0;
2159
2160                 r = bus_poll(bus, false, (uint64_t) -1);
2161                 if (r < 0)
2162                         return r;
2163         }
2164 }
2165
2166 int sd_bus_add_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
2167         struct filter_callback *f;
2168
2169         assert_return(bus, -EINVAL);
2170         assert_return(callback, -EINVAL);
2171         assert_return(!bus_pid_changed(bus), -ECHILD);
2172
2173         f = new0(struct filter_callback, 1);
2174         if (!f)
2175                 return -ENOMEM;
2176         f->callback = callback;
2177         f->userdata = userdata;
2178
2179         bus->filter_callbacks_modified = true;
2180         LIST_PREPEND(callbacks, bus->filter_callbacks, f);
2181         return 0;
2182 }
2183
2184 int sd_bus_remove_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
2185         struct filter_callback *f;
2186
2187         assert_return(bus, -EINVAL);
2188         assert_return(callback, -EINVAL);
2189         assert_return(!bus_pid_changed(bus), -ECHILD);
2190
2191         LIST_FOREACH(callbacks, f, bus->filter_callbacks) {
2192                 if (f->callback == callback && f->userdata == userdata) {
2193                         bus->filter_callbacks_modified = true;
2194                         LIST_REMOVE(callbacks, bus->filter_callbacks, f);
2195                         free(f);
2196                         return 1;
2197                 }
2198         }
2199
2200         return 0;
2201 }
2202
2203 int sd_bus_add_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
2204         struct bus_match_component *components = NULL;
2205         unsigned n_components = 0;
2206         uint64_t cookie = 0;
2207         int r = 0;
2208
2209         assert_return(bus, -EINVAL);
2210         assert_return(match, -EINVAL);
2211         assert_return(!bus_pid_changed(bus), -ECHILD);
2212
2213         r = bus_match_parse(match, &components, &n_components);
2214         if (r < 0)
2215                 goto finish;
2216
2217         if (bus->bus_client) {
2218                 cookie = ++bus->match_cookie;
2219
2220                 r = bus_add_match_internal(bus, match, components, n_components, cookie);
2221                 if (r < 0)
2222                         goto finish;
2223         }
2224
2225         bus->match_callbacks_modified = true;
2226         r = bus_match_add(&bus->match_callbacks, components, n_components, callback, userdata, cookie, NULL);
2227         if (r < 0) {
2228                 if (bus->bus_client)
2229                         bus_remove_match_internal(bus, match, cookie);
2230         }
2231
2232 finish:
2233         bus_match_parse_free(components, n_components);
2234         return r;
2235 }
2236
2237 int sd_bus_remove_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
2238         struct bus_match_component *components = NULL;
2239         unsigned n_components = 0;
2240         int r = 0, q = 0;
2241         uint64_t cookie = 0;
2242
2243         assert_return(bus, -EINVAL);
2244         assert_return(match, -EINVAL);
2245         assert_return(!bus_pid_changed(bus), -ECHILD);
2246
2247         r = bus_match_parse(match, &components, &n_components);
2248         if (r < 0)
2249                 return r;
2250
2251         bus->match_callbacks_modified = true;
2252         r = bus_match_remove(&bus->match_callbacks, components, n_components, callback, userdata, &cookie);
2253
2254         if (bus->bus_client)
2255                 q = bus_remove_match_internal(bus, match, cookie);
2256
2257         bus_match_parse_free(components, n_components);
2258
2259         return r < 0 ? r : q;
2260 }
2261
2262 bool bus_pid_changed(sd_bus *bus) {
2263         assert(bus);
2264
2265         /* We don't support people creating a bus connection and
2266          * keeping it around over a fork(). Let's complain. */
2267
2268         return bus->original_pid != getpid();
2269 }
2270
2271 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
2272         void *bus = userdata;
2273         int r;
2274
2275         assert(bus);
2276
2277         r = sd_bus_process(bus, NULL);
2278         if (r < 0)
2279                 return r;
2280
2281         return 1;
2282 }
2283
2284 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
2285         void *bus = userdata;
2286         int r;
2287
2288         assert(bus);
2289
2290         r = sd_bus_process(bus, NULL);
2291         if (r < 0)
2292                 return r;
2293
2294         return 1;
2295 }
2296
2297 static int prepare_callback(sd_event_source *s, void *userdata) {
2298         sd_bus *bus = userdata;
2299         int r, e;
2300         usec_t until;
2301
2302         assert(s);
2303         assert(bus);
2304
2305         e = sd_bus_get_events(bus);
2306         if (e < 0)
2307                 return e;
2308
2309         if (bus->output_fd != bus->input_fd) {
2310
2311                 r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
2312                 if (r < 0)
2313                         return r;
2314
2315                 r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
2316                 if (r < 0)
2317                         return r;
2318         } else {
2319                 r = sd_event_source_set_io_events(bus->input_io_event_source, e);
2320                 if (r < 0)
2321                         return r;
2322         }
2323
2324         r = sd_bus_get_timeout(bus, &until);
2325         if (r < 0)
2326                 return r;
2327         if (r > 0) {
2328                 int j;
2329
2330                 j = sd_event_source_set_time(bus->time_event_source, until);
2331                 if (j < 0)
2332                         return j;
2333         }
2334
2335         r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
2336         if (r < 0)
2337                 return r;
2338
2339         return 1;
2340 }
2341
2342 static int quit_callback(sd_event_source *event, void *userdata) {
2343         sd_bus *bus = userdata;
2344
2345         assert(event);
2346
2347         sd_bus_flush(bus);
2348
2349         return 1;
2350 }
2351
2352 int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
2353         int r;
2354
2355         assert_return(bus, -EINVAL);
2356         assert_return(event, -EINVAL);
2357         assert_return(!bus->event, -EBUSY);
2358
2359         assert(!bus->input_io_event_source);
2360         assert(!bus->output_io_event_source);
2361         assert(!bus->time_event_source);
2362
2363         bus->event = sd_event_ref(event);
2364
2365         r = sd_event_add_io(event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
2366         if (r < 0)
2367                 goto fail;
2368
2369         r = sd_event_source_set_priority(bus->input_io_event_source, priority);
2370         if (r < 0)
2371                 goto fail;
2372
2373         if (bus->output_fd != bus->input_fd) {
2374                 r = sd_event_add_io(event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
2375                 if (r < 0)
2376                         goto fail;
2377
2378                 r = sd_event_source_set_priority(bus->output_io_event_source, priority);
2379                 if (r < 0)
2380                         goto fail;
2381         }
2382
2383         r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
2384         if (r < 0)
2385                 goto fail;
2386
2387         r = sd_event_add_monotonic(event, 0, 0, time_callback, bus, &bus->time_event_source);
2388         if (r < 0)
2389                 goto fail;
2390
2391         r = sd_event_source_set_priority(bus->time_event_source, priority);
2392         if (r < 0)
2393                 goto fail;
2394
2395         r = sd_event_add_quit(event, quit_callback, bus, &bus->quit_event_source);
2396         if (r < 0)
2397                 goto fail;
2398
2399         return 0;
2400
2401 fail:
2402         sd_bus_detach_event(bus);
2403         return r;
2404 }
2405
2406 int sd_bus_detach_event(sd_bus *bus) {
2407         assert_return(bus, -EINVAL);
2408         assert_return(bus->event, -ENXIO);
2409
2410         if (bus->input_io_event_source)
2411                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
2412
2413         if (bus->output_io_event_source)
2414                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
2415
2416         if (bus->time_event_source)
2417                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
2418
2419         if (bus->quit_event_source)
2420                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
2421
2422         if (bus->event)
2423                 bus->event = sd_event_unref(bus->event);
2424
2425         return 0;
2426 }