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_reset_queues(sd_bus *b) {
 103         unsigned i;
 104
 105         assert(b);
 106
 107         for (i = 0; i < b->rqueue_size; i++)
 108                 sd_bus_message_unref(b->rqueue[i]);
 109         free(b->rqueue);
 110
 111         for (i = 0; i < b->wqueue_size; i++)
 112                 sd_bus_message_unref(b->wqueue[i]);
 113         free(b->wqueue);
 114
 115         b->rqueue = b->wqueue = NULL;
 116         b->rqueue_size = b->wqueue_size = 0;
 117 }
 118
 119 static void bus_free(sd_bus *b) {
 120         struct filter_callback *f;
 121         struct node *n;
 122
 123         assert(b);
 124
 125         sd_bus_detach_event(b);
 126
 127         bus_close_fds(b);
 128
 129         if (b->kdbus_buffer)
 130                 munmap(b->kdbus_buffer, KDBUS_POOL_SIZE);
 131
 132         free(b->rbuffer);
 133         free(b->unique_name);
 134         free(b->auth_buffer);
 135         free(b->address);
 136         free(b->kernel);
 137         free(b->machine);
 138
 139         free(b->exec_path);
 140         strv_free(b->exec_argv);
 141
 142         close_many(b->fds, b->n_fds);
 143         free(b->fds);
 144
 145         bus_reset_queues(b);
 146
 147         hashmap_free_free(b->reply_callbacks);
 148         prioq_free(b->reply_callbacks_prioq);
 149
 150         while ((f = b->filter_callbacks)) {
 151                 LIST_REMOVE(callbacks, b->filter_callbacks, f);
 152                 free(f);
 153         }
 154
 155         bus_match_free(&b->match_callbacks);
 156
 157         hashmap_free_free(b->vtable_methods);
 158         hashmap_free_free(b->vtable_properties);
 159
 160         while ((n = hashmap_first(b->nodes)))
 161                 bus_node_destroy(b, n);
 162
 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         r->wqueue = new(sd_bus_message*, 1);
 194         if (!r->wqueue) {
 195                 free(r);
 196                 return -ENOMEM;
 197         }
 198
 199         *ret = r;
 200         return 0;
 201 }
 202
 203 _public_ int sd_bus_set_address(sd_bus *bus, const char *address) {
 204         char *a;
 205
 206         assert_return(bus, -EINVAL);
 207         assert_return(bus->state == BUS_UNSET, -EPERM);
 208         assert_return(address, -EINVAL);
 209         assert_return(!bus_pid_changed(bus), -ECHILD);
 210
 211         a = strdup(address);
 212         if (!a)
 213                 return -ENOMEM;
 214
 215         free(bus->address);
 216         bus->address = a;
 217
 218         return 0;
 219 }
 220
 221 _public_ int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
 222         assert_return(bus, -EINVAL);
 223         assert_return(bus->state == BUS_UNSET, -EPERM);
 224         assert_return(input_fd >= 0, -EINVAL);
 225         assert_return(output_fd >= 0, -EINVAL);
 226         assert_return(!bus_pid_changed(bus), -ECHILD);
 227
 228         bus->input_fd = input_fd;
 229         bus->output_fd = output_fd;
 230         return 0;
 231 }
 232
 233 _public_ int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
 234         char *p, **a;
 235
 236         assert_return(bus, -EINVAL);
 237         assert_return(bus->state == BUS_UNSET, -EPERM);
 238         assert_return(path, -EINVAL);
 239         assert_return(!strv_isempty(argv), -EINVAL);
 240         assert_return(!bus_pid_changed(bus), -ECHILD);
 241
 242         p = strdup(path);
 243         if (!p)
 244                 return -ENOMEM;
 245
 246         a = strv_copy(argv);
 247         if (!a) {
 248                 free(p);
 249                 return -ENOMEM;
 250         }
 251
 252         free(bus->exec_path);
 253         strv_free(bus->exec_argv);
 254
 255         bus->exec_path = p;
 256         bus->exec_argv = a;
 257
 258         return 0;
 259 }
 260
 261 _public_ int sd_bus_set_bus_client(sd_bus *bus, int b) {
 262         assert_return(bus, -EINVAL);
 263         assert_return(bus->state == BUS_UNSET, -EPERM);
 264         assert_return(!bus_pid_changed(bus), -ECHILD);
 265
 266         bus->bus_client = !!b;
 267         return 0;
 268 }
 269
 270 _public_ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
 271         assert_return(bus, -EINVAL);
 272         assert_return(bus->state == BUS_UNSET, -EPERM);
 273         assert_return(!bus_pid_changed(bus), -ECHILD);
 274
 275         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ACCEPT_FD, b);
 276         return 0;
 277 }
 278
 279 _public_ int sd_bus_negotiate_attach_timestamp(sd_bus *bus, int b) {
 280         assert_return(bus, -EINVAL);
 281         assert_return(bus->state == BUS_UNSET, -EPERM);
 282         assert_return(!bus_pid_changed(bus), -ECHILD);
 283
 284         SET_FLAG(bus->attach_flags, KDBUS_ATTACH_TIMESTAMP, b);
 285         return 0;
 286 }
 287
 288 _public_ int sd_bus_negotiate_attach_creds(sd_bus *bus, uint64_t mask) {
 289         assert_return(bus, -EINVAL);
 290         assert_return(mask <= _SD_BUS_CREDS_MAX, -EINVAL);
 291         assert_return(bus->state == BUS_UNSET, -EPERM);
 292         assert_return(!bus_pid_changed(bus), -ECHILD);
 293
 294         /* The well knowns we need unconditionally, so that matches can work */
 295         bus->creds_mask = mask | SD_BUS_CREDS_WELL_KNOWN_NAMES|SD_BUS_CREDS_UNIQUE_NAME;
 296
 297         return kdbus_translate_attach_flags(bus->creds_mask, &bus->creds_mask);
 298 }
 299
 300 _public_ int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
 301         assert_return(bus, -EINVAL);
 302         assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
 303         assert_return(bus->state == BUS_UNSET, -EPERM);
 304         assert_return(!bus_pid_changed(bus), -ECHILD);
 305
 306         bus->is_server = !!b;
 307         bus->server_id = server_id;
 308         return 0;
 309 }
 310
 311 _public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
 312         assert_return(bus, -EINVAL);
 313         assert_return(bus->state == BUS_UNSET, -EPERM);
 314         assert_return(!bus_pid_changed(bus), -ECHILD);
 315
 316         bus->anonymous_auth = !!b;
 317         return 0;
 318 }
 319
 320 static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata, sd_bus_error *error) {
 321         const char *s;
 322         int r;
 323
 324         assert(bus);
 325         assert(bus->state == BUS_HELLO || bus->state == BUS_CLOSING);
 326         assert(reply);
 327
 328         r = sd_bus_message_get_errno(reply);
 329         if (r < 0)
 330                 return r;
 331         if (r > 0)
 332                 return -r;
 333
 334         r = sd_bus_message_read(reply, "s", &s);
 335         if (r < 0)
 336                 return r;
 337
 338         if (!service_name_is_valid(s) || s[0] != ':')
 339                 return -EBADMSG;
 340
 341         bus->unique_name = strdup(s);
 342         if (!bus->unique_name)
 343                 return -ENOMEM;
 344
 345         if (bus->state == BUS_HELLO)
 346                 bus->state = BUS_RUNNING;
 347
 348         return 1;
 349 }
 350
 351 static int bus_send_hello(sd_bus *bus) {
 352         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
 353         int r;
 354
 355         assert(bus);
 356
 357         if (!bus->bus_client || bus->is_kernel)
 358                 return 0;
 359
 360         r = sd_bus_message_new_method_call(
 361                         bus,
 362                         "org.freedesktop.DBus",
 363                         "/",
 364                         "org.freedesktop.DBus",
 365                         "Hello",
 366                         &m);
 367         if (r < 0)
 368                 return r;
 369
 370         return sd_bus_call_async(bus, m, hello_callback, NULL, 0, &bus->hello_serial);
 371 }
 372
 373 int bus_start_running(sd_bus *bus) {
 374         assert(bus);
 375
 376         if (bus->bus_client && !bus->is_kernel) {
 377                 bus->state = BUS_HELLO;
 378                 return 1;
 379         }
 380
 381         bus->state = BUS_RUNNING;
 382         return 1;
 383 }
 384
 385 static int parse_address_key(const char **p, const char *key, char **value) {
 386         size_t l, n = 0;
 387         const char *a;
 388         char *r = NULL;
 389
 390         assert(p);
 391         assert(*p);
 392         assert(value);
 393
 394         if (key) {
 395                 l = strlen(key);
 396                 if (strncmp(*p, key, l) != 0)
 397                         return 0;
 398
 399                 if ((*p)[l] != '=')
 400                         return 0;
 401
 402                 if (*value)
 403                         return -EINVAL;
 404
 405                 a = *p + l + 1;
 406         } else
 407                 a = *p;
 408
 409         while (*a != ';' && *a != ',' && *a != 0) {
 410                 char c, *t;
 411
 412                 if (*a == '%') {
 413                         int x, y;
 414
 415                         x = unhexchar(a[1]);
 416                         if (x < 0) {
 417                                 free(r);
 418                                 return x;
 419                         }
 420
 421                         y = unhexchar(a[2]);
 422                         if (y < 0) {
 423                                 free(r);
 424                                 return y;
 425                         }
 426
 427                         c = (char) ((x << 4) | y);
 428                         a += 3;
 429                 } else {
 430                         c = *a;
 431                         a++;
 432                 }
 433
 434                 t = realloc(r, n + 2);
 435                 if (!t) {
 436                         free(r);
 437                         return -ENOMEM;
 438                 }
 439
 440                 r = t;
 441                 r[n++] = c;
 442         }
 443
 444         if (!r) {
 445                 r = strdup("");
 446                 if (!r)
 447                         return -ENOMEM;
 448         } else
 449                 r[n] = 0;
 450
 451         if (*a == ',')
 452                 a++;
 453
 454         *p = a;
 455
 456         free(*value);
 457         *value = r;
 458
 459         return 1;
 460 }
 461
 462 static void skip_address_key(const char **p) {
 463         assert(p);
 464         assert(*p);
 465
 466         *p += strcspn(*p, ",");
 467
 468         if (**p == ',')
 469                 (*p) ++;
 470 }
 471
 472 static int parse_unix_address(sd_bus *b, const char **p, char **guid) {
 473         _cleanup_free_ char *path = NULL, *abstract = NULL;
 474         size_t l;
 475         int r;
 476
 477         assert(b);
 478         assert(p);
 479         assert(*p);
 480         assert(guid);
 481
 482         while (**p != 0 && **p != ';') {
 483                 r = parse_address_key(p, "guid", guid);
 484                 if (r < 0)
 485                         return r;
 486                 else if (r > 0)
 487                         continue;
 488
 489                 r = parse_address_key(p, "path", &path);
 490                 if (r < 0)
 491                         return r;
 492                 else if (r > 0)
 493                         continue;
 494
 495                 r = parse_address_key(p, "abstract", &abstract);
 496                 if (r < 0)
 497                         return r;
 498                 else if (r > 0)
 499                         continue;
 500
 501                 skip_address_key(p);
 502         }
 503
 504         if (!path && !abstract)
 505                 return -EINVAL;
 506
 507         if (path && abstract)
 508                 return -EINVAL;
 509
 510         if (path) {
 511                 l = strlen(path);
 512                 if (l > sizeof(b->sockaddr.un.sun_path))
 513                         return -E2BIG;
 514
 515                 b->sockaddr.un.sun_family = AF_UNIX;
 516                 strncpy(b->sockaddr.un.sun_path, path, sizeof(b->sockaddr.un.sun_path));
 517                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l;
 518         } else if (abstract) {
 519                 l = strlen(abstract);
 520                 if (l > sizeof(b->sockaddr.un.sun_path) - 1)
 521                         return -E2BIG;
 522
 523                 b->sockaddr.un.sun_family = AF_UNIX;
 524                 b->sockaddr.un.sun_path[0] = 0;
 525                 strncpy(b->sockaddr.un.sun_path+1, abstract, sizeof(b->sockaddr.un.sun_path)-1);
 526                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + 1 + l;
 527         }
 528
 529         return 0;
 530 }
 531
 532 static int parse_tcp_address(sd_bus *b, const char **p, char **guid) {
 533         _cleanup_free_ char *host = NULL, *port = NULL, *family = NULL;
 534         int r;
 535         struct addrinfo *result, hints = {
 536                 .ai_socktype = SOCK_STREAM,
 537                 .ai_flags = AI_ADDRCONFIG,
 538         };
 539
 540         assert(b);
 541         assert(p);
 542         assert(*p);
 543         assert(guid);
 544
 545         while (**p != 0 && **p != ';') {
 546                 r = parse_address_key(p, "guid", guid);
 547                 if (r < 0)
 548                         return r;
 549                 else if (r > 0)
 550                         continue;
 551
 552                 r = parse_address_key(p, "host", &host);
 553                 if (r < 0)
 554                         return r;
 555                 else if (r > 0)
 556                         continue;
 557
 558                 r = parse_address_key(p, "port", &port);
 559                 if (r < 0)
 560                         return r;
 561                 else if (r > 0)
 562                         continue;
 563
 564                 r = parse_address_key(p, "family", &family);
 565                 if (r < 0)
 566                         return r;
 567                 else if (r > 0)
 568                         continue;
 569
 570                 skip_address_key(p);
 571         }
 572
 573         if (!host || !port)
 574                 return -EINVAL;
 575
 576         if (family) {
 577                 if (streq(family, "ipv4"))
 578                         hints.ai_family = AF_INET;
 579                 else if (streq(family, "ipv6"))
 580                         hints.ai_family = AF_INET6;
 581                 else
 582                         return -EINVAL;
 583         }
 584
 585         r = getaddrinfo(host, port, &hints, &result);
 586         if (r == EAI_SYSTEM)
 587                 return -errno;
 588         else if (r != 0)
 589                 return -EADDRNOTAVAIL;
 590
 591         memcpy(&b->sockaddr, result->ai_addr, result->ai_addrlen);
 592         b->sockaddr_size = result->ai_addrlen;
 593
 594         freeaddrinfo(result);
 595
 596         return 0;
 597 }
 598
 599 static int parse_exec_address(sd_bus *b, const char **p, char **guid) {
 600         char *path = NULL;
 601         unsigned n_argv = 0, j;
 602         char **argv = NULL;
 603         int r;
 604
 605         assert(b);
 606         assert(p);
 607         assert(*p);
 608         assert(guid);
 609
 610         while (**p != 0 && **p != ';') {
 611                 r = parse_address_key(p, "guid", guid);
 612                 if (r < 0)
 613                         goto fail;
 614                 else if (r > 0)
 615                         continue;
 616
 617                 r = parse_address_key(p, "path", &path);
 618                 if (r < 0)
 619                         goto fail;
 620                 else if (r > 0)
 621                         continue;
 622
 623                 if (startswith(*p, "argv")) {
 624                         unsigned ul;
 625
 626                         errno = 0;
 627                         ul = strtoul(*p + 4, (char**) p, 10);
 628                         if (errno > 0 || **p != '=' || ul > 256) {
 629                                 r = -EINVAL;
 630                                 goto fail;
 631                         }
 632
 633                         (*p) ++;
 634
 635                         if (ul >= n_argv) {
 636                                 char **x;
 637
 638                                 x = realloc(argv, sizeof(char*) * (ul + 2));
 639                                 if (!x) {
 640                                         r = -ENOMEM;
 641                                         goto fail;
 642                                 }
 643
 644                                 memset(x + n_argv, 0, sizeof(char*) * (ul - n_argv + 2));
 645
 646                                 argv = x;
 647                                 n_argv = ul + 1;
 648                         }
 649
 650                         r = parse_address_key(p, NULL, argv + ul);
 651                         if (r < 0)
 652                                 goto fail;
 653
 654                         continue;
 655                 }
 656
 657                 skip_address_key(p);
 658         }
 659
 660         if (!path) {
 661                 r = -EINVAL;
 662                 goto fail;
 663         }
 664
 665         /* Make sure there are no holes in the array, with the
 666          * exception of argv[0] */
 667         for (j = 1; j < n_argv; j++)
 668                 if (!argv[j]) {
 669                         r = -EINVAL;
 670                         goto fail;
 671                 }
 672
 673         if (argv && argv[0] == NULL) {
 674                 argv[0] = strdup(path);
 675                 if (!argv[0]) {
 676                         r = -ENOMEM;
 677                         goto fail;
 678                 }
 679         }
 680
 681         b->exec_path = path;
 682         b->exec_argv = argv;
 683         return 0;
 684
 685 fail:
 686         for (j = 0; j < n_argv; j++)
 687                 free(argv[j]);
 688
 689         free(argv);
 690         free(path);
 691         return r;
 692 }
 693
 694 static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
 695         _cleanup_free_ char *path = NULL;
 696         int r;
 697
 698         assert(b);
 699         assert(p);
 700         assert(*p);
 701         assert(guid);
 702
 703         while (**p != 0 && **p != ';') {
 704                 r = parse_address_key(p, "guid", guid);
 705                 if (r < 0)
 706                         return r;
 707                 else if (r > 0)
 708                         continue;
 709
 710                 r = parse_address_key(p, "path", &path);
 711                 if (r < 0)
 712                         return r;
 713                 else if (r > 0)
 714                         continue;
 715
 716                 skip_address_key(p);
 717         }
 718
 719         if (!path)
 720                 return -EINVAL;
 721
 722         free(b->kernel);
 723         b->kernel = path;
 724         path = NULL;
 725
 726         return 0;
 727 }
 728
 729 static int parse_container_address(sd_bus *b, const char **p, char **guid) {
 730         _cleanup_free_ char *machine = NULL;
 731         int r;
 732
 733         assert(b);
 734         assert(p);
 735         assert(*p);
 736         assert(guid);
 737
 738         while (**p != 0 && **p != ';') {
 739                 r = parse_address_key(p, "guid", guid);
 740                 if (r < 0)
 741                         return r;
 742                 else if (r > 0)
 743                         continue;
 744
 745                 r = parse_address_key(p, "machine", &machine);
 746                 if (r < 0)
 747                         return r;
 748                 else if (r > 0)
 749                         continue;
 750
 751                 skip_address_key(p);
 752         }
 753
 754         if (!machine)
 755                 return -EINVAL;
 756
 757         free(b->machine);
 758         b->machine = machine;
 759         machine = NULL;
 760
 761         b->sockaddr.un.sun_family = AF_UNIX;
 762         strncpy(b->sockaddr.un.sun_path, "/var/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
 763         b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/var/run/dbus/system_bus_socket") - 1;
 764
 765         return 0;
 766 }
 767
 768 static void bus_reset_parsed_address(sd_bus *b) {
 769         assert(b);
 770
 771         zero(b->sockaddr);
 772         b->sockaddr_size = 0;
 773         strv_free(b->exec_argv);
 774         free(b->exec_path);
 775         b->exec_path = NULL;
 776         b->exec_argv = NULL;
 777         b->server_id = SD_ID128_NULL;
 778         free(b->kernel);
 779         b->kernel = NULL;
 780         free(b->machine);
 781         b->machine = NULL;
 782 }
 783
 784 static int bus_parse_next_address(sd_bus *b) {
 785         _cleanup_free_ char *guid = NULL;
 786         const char *a;
 787         int r;
 788
 789         assert(b);
 790
 791         if (!b->address)
 792                 return 0;
 793         if (b->address[b->address_index] == 0)
 794                 return 0;
 795
 796         bus_reset_parsed_address(b);
 797
 798         a = b->address + b->address_index;
 799
 800         while (*a != 0) {
 801
 802                 if (*a == ';') {
 803                         a++;
 804                         continue;
 805                 }
 806
 807                 if (startswith(a, "unix:")) {
 808                         a += 5;
 809
 810                         r = parse_unix_address(b, &a, &guid);
 811                         if (r < 0)
 812                                 return r;
 813                         break;
 814
 815                 } else if (startswith(a, "tcp:")) {
 816
 817                         a += 4;
 818                         r = parse_tcp_address(b, &a, &guid);
 819                         if (r < 0)
 820                                 return r;
 821
 822                         break;
 823
 824                 } else if (startswith(a, "unixexec:")) {
 825
 826                         a += 9;
 827                         r = parse_exec_address(b, &a, &guid);
 828                         if (r < 0)
 829                                 return r;
 830
 831                         break;
 832
 833                 } else if (startswith(a, "kernel:")) {
 834
 835                         a += 7;
 836                         r = parse_kernel_address(b, &a, &guid);
 837                         if (r < 0)
 838                                 return r;
 839
 840                         break;
 841                 } else if (startswith(a, "x-container:")) {
 842
 843                         a += 12;
 844                         r = parse_container_address(b, &a, &guid);
 845                         if (r < 0)
 846                                 return r;
 847
 848                         break;
 849                 }
 850
 851                 a = strchr(a, ';');
 852                 if (!a)
 853                         return 0;
 854         }
 855
 856         if (guid) {
 857                 r = sd_id128_from_string(guid, &b->server_id);
 858                 if (r < 0)
 859                         return r;
 860         }
 861
 862         b->address_index = a - b->address;
 863         return 1;
 864 }
 865
 866 static int bus_start_address(sd_bus *b) {
 867         int r;
 868
 869         assert(b);
 870
 871         for (;;) {
 872                 sd_bus_close(b);
 873
 874                 if (b->exec_path) {
 875
 876                         r = bus_socket_exec(b);
 877                         if (r >= 0)
 878                                 return r;
 879
 880                         b->last_connect_error = -r;
 881                 } else if (b->kernel) {
 882
 883                         r = bus_kernel_connect(b);
 884                         if (r >= 0)
 885                                 return r;
 886
 887                         b->last_connect_error = -r;
 888
 889                 } else if (b->machine) {
 890
 891                         r = bus_container_connect(b);
 892                         if (r >= 0)
 893                                 return r;
 894
 895                         b->last_connect_error = -r;
 896
 897                 } else if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
 898
 899                         r = bus_socket_connect(b);
 900                         if (r >= 0)
 901                                 return r;
 902
 903                         b->last_connect_error = -r;
 904                 }
 905
 906                 r = bus_parse_next_address(b);
 907                 if (r < 0)
 908                         return r;
 909                 if (r == 0)
 910                         return b->last_connect_error ? -b->last_connect_error : -ECONNREFUSED;
 911         }
 912 }
 913
 914 int bus_next_address(sd_bus *b) {
 915         assert(b);
 916
 917         bus_reset_parsed_address(b);
 918         return bus_start_address(b);
 919 }
 920
 921 static int bus_start_fd(sd_bus *b) {
 922         struct stat st;
 923         int r;
 924
 925         assert(b);
 926         assert(b->input_fd >= 0);
 927         assert(b->output_fd >= 0);
 928
 929         r = fd_nonblock(b->input_fd, true);
 930         if (r < 0)
 931                 return r;
 932
 933         r = fd_cloexec(b->input_fd, true);
 934         if (r < 0)
 935                 return r;
 936
 937         if (b->input_fd != b->output_fd) {
 938                 r = fd_nonblock(b->output_fd, true);
 939                 if (r < 0)
 940                         return r;
 941
 942                 r = fd_cloexec(b->output_fd, true);
 943                 if (r < 0)
 944                         return r;
 945         }
 946
 947         if (fstat(b->input_fd, &st) < 0)
 948                 return -errno;
 949
 950         if (S_ISCHR(b->input_fd))
 951                 return bus_kernel_take_fd(b);
 952         else
 953                 return bus_socket_take_fd(b);
 954 }
 955
 956 _public_ int sd_bus_start(sd_bus *bus) {
 957         int r;
 958
 959         assert_return(bus, -EINVAL);
 960         assert_return(bus->state == BUS_UNSET, -EPERM);
 961         assert_return(!bus_pid_changed(bus), -ECHILD);
 962
 963         bus->state = BUS_OPENING;
 964
 965         if (bus->is_server && bus->bus_client)
 966                 return -EINVAL;
 967
 968         if (bus->input_fd >= 0)
 969                 r = bus_start_fd(bus);
 970         else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel || bus->machine)
 971                 r = bus_start_address(bus);
 972         else
 973                 return -EINVAL;
 974
 975         if (r < 0)
 976                 return r;
 977
 978         return bus_send_hello(bus);
 979 }
 980
 981 _public_ int sd_bus_open_system(sd_bus **ret) {
 982         const char *e;
 983         sd_bus *b;
 984         int r;
 985
 986         assert_return(ret, -EINVAL);
 987
 988         r = sd_bus_new(&b);
 989         if (r < 0)
 990                 return r;
 991
 992         e = secure_getenv("DBUS_SYSTEM_BUS_ADDRESS");
 993         if (e)
 994                 r = sd_bus_set_address(b, e);
 995         else
 996 #ifdef ENABLE_KDBUS
 997                 r = sd_bus_set_address(b, "kernel:path=/dev/kdbus/0-system/bus;unix:path=/run/dbus/system_bus_socket");
 998 #else
 999                 r = sd_bus_set_address(b, "unix:path=/run/dbus/system_bus_socket");
1000 #endif
1001
1002         if (r < 0)
1003                 goto fail;
1004
1005         b->bus_client = true;
1006
1007         r = sd_bus_start(b);
1008         if (r < 0)
1009                 goto fail;
1010
1011         *ret = b;
1012         return 0;
1013
1014 fail:
1015         bus_free(b);
1016         return r;
1017 }
1018
1019 _public_ int sd_bus_open_user(sd_bus **ret) {
1020         const char *e;
1021         sd_bus *b;
1022         int r;
1023
1024         assert_return(ret, -EINVAL);
1025
1026         r = sd_bus_new(&b);
1027         if (r < 0)
1028                 return r;
1029
1030         e = secure_getenv("DBUS_SESSION_BUS_ADDRESS");
1031         if (e) {
1032                 r = sd_bus_set_address(b, e);
1033                 if (r < 0)
1034                         goto fail;
1035         } else {
1036                 e = secure_getenv("XDG_RUNTIME_DIR");
1037                 if (e) {
1038                         _cleanup_free_ char *ee = NULL;
1039
1040                         ee = bus_address_escape(e);
1041                         if (!ee) {
1042                                 r = -ENOMEM;
1043                                 goto fail;
1044                         }
1045
1046 #ifdef ENABLE_KDBUS
1047                         asprintf(&b->address, "kernel:path=/dev/kdbus/%lu-user/bus;unix:path=%s/bus", (unsigned long) getuid(), ee);
1048 #else
1049                         b->address = strjoin("unix:path=", ee, "/bus", NULL);
1050 #endif
1051                 } else {
1052 #ifdef ENABLE_KDBUS
1053                         asprintf(&b->address, "kernel:path=/dev/kdbus/%lu-user/bus", (unsigned long) getuid());
1054 #else
1055                         return -ECONNREFUSED;
1056 #endif
1057                 }
1058
1059                 if (!b->address) {
1060                         r = -ENOMEM;
1061                         goto fail;
1062                 }
1063         }
1064
1065         b->bus_client = true;
1066
1067         r = sd_bus_start(b);
1068         if (r < 0)
1069                 goto fail;
1070
1071         *ret = b;
1072         return 0;
1073
1074 fail:
1075         bus_free(b);
1076         return r;
1077 }
1078
1079 _public_ int sd_bus_open_system_remote(const char *host, sd_bus **ret) {
1080         _cleanup_free_ char *e = NULL;
1081         char *p = NULL;
1082         sd_bus *bus;
1083         int r;
1084
1085         assert_return(host, -EINVAL);
1086         assert_return(ret, -EINVAL);
1087
1088         e = bus_address_escape(host);
1089         if (!e)
1090                 return -ENOMEM;
1091
1092         p = strjoin("unixexec:path=ssh,argv1=-xT,argv2=", e, ",argv3=systemd-stdio-bridge", NULL);
1093         if (!p)
1094                 return -ENOMEM;
1095
1096         r = sd_bus_new(&bus);
1097         if (r < 0) {
1098                 free(p);
1099                 return r;
1100         }
1101
1102         bus->address = p;
1103         bus->bus_client = true;
1104
1105         r = sd_bus_start(bus);
1106         if (r < 0) {
1107                 bus_free(bus);
1108                 return r;
1109         }
1110
1111         *ret = bus;
1112         return 0;
1113 }
1114
1115 _public_ int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
1116         _cleanup_free_ char *e = NULL;
1117         sd_bus *bus;
1118         char *p;
1119         int r;
1120
1121         assert_return(machine, -EINVAL);
1122         assert_return(ret, -EINVAL);
1123
1124         e = bus_address_escape(machine);
1125         if (!e)
1126                 return -ENOMEM;
1127
1128         p = strjoin("x-container:machine=", e, NULL);
1129         if (!p)
1130                 return -ENOMEM;
1131
1132         r = sd_bus_new(&bus);
1133         if (r < 0) {
1134                 free(p);
1135                 return r;
1136         }
1137
1138         bus->address = p;
1139         bus->bus_client = true;
1140
1141         r = sd_bus_start(bus);
1142         if (r < 0) {
1143                 bus_free(bus);
1144                 return r;
1145         }
1146
1147         *ret = bus;
1148         return 0;
1149 }
1150
1151 _public_ void sd_bus_close(sd_bus *bus) {
1152
1153         if (!bus)
1154                 return;
1155         if (bus->state == BUS_CLOSED)
1156                 return;
1157         if (bus_pid_changed(bus))
1158                 return;
1159
1160         bus->state = BUS_CLOSED;
1161
1162         sd_bus_detach_event(bus);
1163
1164         /* Drop all queued messages so that they drop references to
1165          * the bus object and the bus may be freed */
1166         bus_reset_queues(bus);
1167
1168         if (!bus->is_kernel)
1169                 bus_close_fds(bus);
1170
1171         /* We'll leave the fd open in case this is a kernel bus, since
1172          * there might still be memblocks around that reference this
1173          * bus, and they might need to invoke the * KDBUS_CMD_FREE
1174          * ioctl on the fd when they are freed. */
1175 }
1176
1177 static void bus_enter_closing(sd_bus *bus) {
1178         assert(bus);
1179
1180         if (bus->state != BUS_OPENING &&
1181             bus->state != BUS_AUTHENTICATING &&
1182             bus->state != BUS_HELLO &&
1183             bus->state != BUS_RUNNING)
1184                 return;
1185
1186         bus->state = BUS_CLOSING;
1187 }
1188
1189 _public_ sd_bus *sd_bus_ref(sd_bus *bus) {
1190         assert_return(bus, NULL);
1191
1192         assert_se(REFCNT_INC(bus->n_ref) >= 2);
1193
1194         return bus;
1195 }
1196
1197 _public_ sd_bus *sd_bus_unref(sd_bus *bus) {
1198         assert_return(bus, NULL);
1199
1200         if (REFCNT_DEC(bus->n_ref) <= 0)
1201                 bus_free(bus);
1202
1203         return NULL;
1204 }
1205
1206 _public_ int sd_bus_is_open(sd_bus *bus) {
1207
1208         assert_return(bus, -EINVAL);
1209         assert_return(!bus_pid_changed(bus), -ECHILD);
1210
1211         return BUS_IS_OPEN(bus->state);
1212 }
1213
1214 _public_ int sd_bus_can_send(sd_bus *bus, char type) {
1215         int r;
1216
1217         assert_return(bus, -EINVAL);
1218         assert_return(bus->state != BUS_UNSET, -ENOTCONN);
1219         assert_return(!bus_pid_changed(bus), -ECHILD);
1220
1221         if (type == SD_BUS_TYPE_UNIX_FD) {
1222                 if (!(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD))
1223                         return 0;
1224
1225                 r = bus_ensure_running(bus);
1226                 if (r < 0)
1227                         return r;
1228
1229                 return bus->can_fds;
1230         }
1231
1232         return bus_type_is_valid(type);
1233 }
1234
1235 _public_ int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
1236         int r;
1237
1238         assert_return(bus, -EINVAL);
1239         assert_return(server_id, -EINVAL);
1240         assert_return(!bus_pid_changed(bus), -ECHILD);
1241
1242         r = bus_ensure_running(bus);
1243         if (r < 0)
1244                 return r;
1245
1246         *server_id = bus->server_id;
1247         return 0;
1248 }
1249
1250 static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
1251         assert(b);
1252         assert(m);
1253
1254         if (m->header->version > b->message_version)
1255                 return -EPERM;
1256
1257         if (m->sealed) {
1258                 /* If we copy the same message to multiple
1259                  * destinations, avoid using the same serial
1260                  * numbers. */
1261                 b->serial = MAX(b->serial, BUS_MESSAGE_SERIAL(m));
1262                 return 0;
1263         }
1264
1265         return bus_message_seal(m, ++b->serial);
1266 }
1267
1268 int bus_seal_synthetic_message(sd_bus *b, sd_bus_message *m) {
1269         assert(b);
1270         assert(m);
1271
1272         if (m->header->version > b->message_version)
1273                 return -EPERM;
1274
1275         /* The bus specification says the serial number cannot be 0,
1276          * hence let's fill something in for synthetic messages. Since
1277          * synthetic messages might have a fake sender and we don't
1278          * want to interfere with the real sender's serial numbers we
1279          * pick a fixed, artifical one. We use (uint32_t) -1 rather
1280          * than (uint64_t) -1 since dbus1 only had 32bit identifiers,
1281          * even though kdbus can do 64bit. */
1282
1283         return bus_message_seal(m, 0xFFFFFFFFULL);
1284 }
1285
1286 static int bus_write_message(sd_bus *bus, sd_bus_message *message, size_t *idx) {
1287         int r;
1288
1289         assert(bus);
1290         assert(message);
1291
1292         if (bus->is_kernel)
1293                 return bus_kernel_write_message(bus, message);
1294         else
1295                 return bus_socket_write_message(bus, message, idx);
1296
1297         return r;
1298 }
1299
1300 static int dispatch_wqueue(sd_bus *bus) {
1301         int r, ret = 0;
1302
1303         assert(bus);
1304         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1305
1306         while (bus->wqueue_size > 0) {
1307
1308                 r = bus_write_message(bus, bus->wqueue[0], &bus->windex);
1309                 if (r < 0)
1310                         return r;
1311                 else if (r == 0)
1312                         /* Didn't do anything this time */
1313                         return ret;
1314                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1315                         /* Fully written. Let's drop the entry from
1316                          * the queue.
1317                          *
1318                          * This isn't particularly optimized, but
1319                          * well, this is supposed to be our worst-case
1320                          * buffer only, and the socket buffer is
1321                          * supposed to be our primary buffer, and if
1322                          * it got full, then all bets are off
1323                          * anyway. */
1324
1325                         sd_bus_message_unref(bus->wqueue[0]);
1326                         bus->wqueue_size --;
1327                         memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1328                         bus->windex = 0;
1329
1330                         ret = 1;
1331                 }
1332         }
1333
1334         return ret;
1335 }
1336
1337 static int bus_read_message(sd_bus *bus) {
1338         assert(bus);
1339
1340         if (bus->is_kernel)
1341                 return bus_kernel_read_message(bus);
1342         else
1343                 return bus_socket_read_message(bus);
1344 }
1345
1346 int bus_rqueue_make_room(sd_bus *bus) {
1347         sd_bus_message **q;
1348         unsigned x;
1349
1350         x = bus->rqueue_size + 1;
1351
1352         if (bus->rqueue_allocated >= x)
1353                 return 0;
1354
1355         if (x > BUS_RQUEUE_MAX)
1356                 return -ENOBUFS;
1357
1358         q = realloc(bus->rqueue, x * sizeof(sd_bus_message*));
1359         if (!q)
1360                 return -ENOMEM;
1361
1362         bus->rqueue = q;
1363         bus->rqueue_allocated = x;
1364
1365         return 0;
1366 }
1367
1368 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
1369         int r, ret = 0;
1370
1371         assert(bus);
1372         assert(m);
1373         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1374
1375         for (;;) {
1376                 if (bus->rqueue_size > 0) {
1377                         /* Dispatch a queued message */
1378
1379                         *m = bus->rqueue[0];
1380                         bus->rqueue_size --;
1381                         memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1382                         return 1;
1383                 }
1384
1385                 /* Try to read a new message */
1386                 r = bus_read_message(bus);
1387                 if (r < 0)
1388                         return r;
1389                 if (r == 0)
1390                         return ret;
1391
1392                 ret = 1;
1393         }
1394 }
1395
1396 _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
1397         int r;
1398
1399         assert_return(bus, -EINVAL);
1400         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1401         assert_return(m, -EINVAL);
1402         assert_return(!bus_pid_changed(bus), -ECHILD);
1403
1404         if (m->n_fds > 0) {
1405                 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1406                 if (r < 0)
1407                         return r;
1408                 if (r == 0)
1409                         return -ENOTSUP;
1410         }
1411
1412         /* If the serial number isn't kept, then we know that no reply
1413          * is expected */
1414         if (!serial && !m->sealed)
1415                 m->header->flags |= SD_BUS_MESSAGE_NO_REPLY_EXPECTED;
1416
1417         r = bus_seal_message(bus, m);
1418         if (r < 0)
1419                 return r;
1420
1421         /* If this is a reply and no reply was requested, then let's
1422          * suppress this, if we can */
1423         if (m->dont_send && !serial)
1424                 return 1;
1425
1426         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1427                 size_t idx = 0;
1428
1429                 r = bus_write_message(bus, m, &idx);
1430                 if (r < 0) {
1431                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1432                                 bus_enter_closing(bus);
1433
1434                         return r;
1435                 } else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
1436                         /* Wasn't fully written. So let's remember how
1437                          * much was written. Note that the first entry
1438                          * of the wqueue array is always allocated so
1439                          * that we always can remember how much was
1440                          * written. */
1441                         bus->wqueue[0] = sd_bus_message_ref(m);
1442                         bus->wqueue_size = 1;
1443                         bus->windex = idx;
1444                 }
1445         } else {
1446                 sd_bus_message **q;
1447
1448                 /* Just append it to the queue. */
1449
1450                 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1451                         return -ENOBUFS;
1452
1453                 q = realloc(bus->wqueue, sizeof(sd_bus_message*) * (bus->wqueue_size + 1));
1454                 if (!q)
1455                         return -ENOMEM;
1456
1457                 bus->wqueue = q;
1458                 q[bus->wqueue_size ++] = sd_bus_message_ref(m);
1459         }
1460
1461         if (serial)
1462                 *serial = BUS_MESSAGE_SERIAL(m);
1463
1464         return 1;
1465 }
1466
1467 _public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *serial) {
1468         int r;
1469
1470         assert_return(bus, -EINVAL);
1471         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1472         assert_return(m, -EINVAL);
1473         assert_return(!bus_pid_changed(bus), -ECHILD);
1474
1475         if (!streq_ptr(m->destination, destination)) {
1476
1477                 if (!destination)
1478                         return -EEXIST;
1479
1480                 r = sd_bus_message_set_destination(m, destination);
1481                 if (r < 0)
1482                         return r;
1483         }
1484
1485         return sd_bus_send(bus, m, serial);
1486 }
1487
1488 static usec_t calc_elapse(uint64_t usec) {
1489         if (usec == (uint64_t) -1)
1490                 return 0;
1491
1492         if (usec == 0)
1493                 usec = BUS_DEFAULT_TIMEOUT;
1494
1495         return now(CLOCK_MONOTONIC) + usec;
1496 }
1497
1498 static int timeout_compare(const void *a, const void *b) {
1499         const struct reply_callback *x = a, *y = b;
1500
1501         if (x->timeout != 0 && y->timeout == 0)
1502                 return -1;
1503
1504         if (x->timeout == 0 && y->timeout != 0)
1505                 return 1;
1506
1507         if (x->timeout < y->timeout)
1508                 return -1;
1509
1510         if (x->timeout > y->timeout)
1511                 return 1;
1512
1513         return 0;
1514 }
1515
1516 _public_ int sd_bus_call_async(
1517                 sd_bus *bus,
1518                 sd_bus_message *m,
1519                 sd_bus_message_handler_t callback,
1520                 void *userdata,
1521                 uint64_t usec,
1522                 uint64_t *serial) {
1523
1524         struct reply_callback *c;
1525         int r;
1526
1527         assert_return(bus, -EINVAL);
1528         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1529         assert_return(m, -EINVAL);
1530         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1531         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1532         assert_return(callback, -EINVAL);
1533         assert_return(!bus_pid_changed(bus), -ECHILD);
1534
1535         r = hashmap_ensure_allocated(&bus->reply_callbacks, uint64_hash_func, uint64_compare_func);
1536         if (r < 0)
1537                 return r;
1538
1539         if (usec != (uint64_t) -1) {
1540                 r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1541                 if (r < 0)
1542                         return r;
1543         }
1544
1545         r = bus_seal_message(bus, m);
1546         if (r < 0)
1547                 return r;
1548
1549         c = new0(struct reply_callback, 1);
1550         if (!c)
1551                 return -ENOMEM;
1552
1553         c->callback = callback;
1554         c->userdata = userdata;
1555         c->serial = BUS_MESSAGE_SERIAL(m);
1556         c->timeout = calc_elapse(usec);
1557
1558         r = hashmap_put(bus->reply_callbacks, &c->serial, c);
1559         if (r < 0) {
1560                 free(c);
1561                 return r;
1562         }
1563
1564         if (c->timeout != 0) {
1565                 r = prioq_put(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1566                 if (r < 0) {
1567                         c->timeout = 0;
1568                         sd_bus_call_async_cancel(bus, c->serial);
1569                         return r;
1570                 }
1571         }
1572
1573         r = sd_bus_send(bus, m, serial);
1574         if (r < 0) {
1575                 sd_bus_call_async_cancel(bus, c->serial);
1576                 return r;
1577         }
1578
1579         return r;
1580 }
1581
1582 _public_ int sd_bus_call_async_cancel(sd_bus *bus, uint64_t serial) {
1583         struct reply_callback *c;
1584
1585         assert_return(bus, -EINVAL);
1586         assert_return(serial != 0, -EINVAL);
1587         assert_return(!bus_pid_changed(bus), -ECHILD);
1588
1589         c = hashmap_remove(bus->reply_callbacks, &serial);
1590         if (!c)
1591                 return 0;
1592
1593         if (c->timeout != 0)
1594                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1595
1596         free(c);
1597         return 1;
1598 }
1599
1600 int bus_ensure_running(sd_bus *bus) {
1601         int r;
1602
1603         assert(bus);
1604
1605         if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
1606                 return -ENOTCONN;
1607         if (bus->state == BUS_RUNNING)
1608                 return 1;
1609
1610         for (;;) {
1611                 r = sd_bus_process(bus, NULL);
1612                 if (r < 0)
1613                         return r;
1614                 if (bus->state == BUS_RUNNING)
1615                         return 1;
1616                 if (r > 0)
1617                         continue;
1618
1619                 r = sd_bus_wait(bus, (uint64_t) -1);
1620                 if (r < 0)
1621                         return r;
1622         }
1623 }
1624
1625 _public_ int sd_bus_call(
1626                 sd_bus *bus,
1627                 sd_bus_message *m,
1628                 uint64_t usec,
1629                 sd_bus_error *error,
1630                 sd_bus_message **reply) {
1631
1632         usec_t timeout;
1633         uint64_t serial;
1634         unsigned i;
1635         int r;
1636
1637         assert_return(bus, -EINVAL);
1638         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1639         assert_return(m, -EINVAL);
1640         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1641         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1642         assert_return(!bus_error_is_dirty(error), -EINVAL);
1643         assert_return(!bus_pid_changed(bus), -ECHILD);
1644
1645         r = bus_ensure_running(bus);
1646         if (r < 0)
1647                 return r;
1648
1649         i = bus->rqueue_size;
1650
1651         r = sd_bus_send(bus, m, &serial);
1652         if (r < 0)
1653                 return r;
1654
1655         timeout = calc_elapse(usec);
1656
1657         for (;;) {
1658                 usec_t left;
1659
1660                 while (i < bus->rqueue_size) {
1661                         sd_bus_message *incoming = NULL;
1662
1663                         incoming = bus->rqueue[i];
1664
1665                         if (incoming->reply_serial == serial) {
1666                                 /* Found a match! */
1667
1668                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
1669                                 bus->rqueue_size--;
1670
1671                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
1672
1673                                         if (reply)
1674                                                 *reply = incoming;
1675                                         else
1676                                                 sd_bus_message_unref(incoming);
1677
1678                                         return 1;
1679                                 } else if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR)
1680                                         r = sd_bus_error_copy(error, &incoming->error);
1681                                 else
1682                                         r = -EIO;
1683
1684                                 sd_bus_message_unref(incoming);
1685                                 return r;
1686
1687                         } else if (incoming->header->serial == serial &&
1688                                    bus->unique_name &&
1689                                    incoming->sender &&
1690                                    streq(bus->unique_name, incoming->sender)) {
1691
1692                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
1693                                 bus->rqueue_size--;
1694
1695                                 /* Our own message? Somebody is trying
1696                                  * to send its own client a message,
1697                                  * let's not dead-lock, let's fail
1698                                  * immediately. */
1699
1700                                 sd_bus_message_unref(incoming);
1701                                 return -ELOOP;
1702                         }
1703
1704                         /* Try to read more, right-away */
1705                         i++;
1706                 }
1707
1708                 r = bus_read_message(bus);
1709                 if (r < 0) {
1710                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1711                                 bus_enter_closing(bus);
1712
1713                         return r;
1714                 }
1715                 if (r > 0)
1716                         continue;
1717
1718                 if (timeout > 0) {
1719                         usec_t n;
1720
1721                         n = now(CLOCK_MONOTONIC);
1722                         if (n >= timeout)
1723                                 return -ETIMEDOUT;
1724
1725                         left = timeout - n;
1726                 } else
1727                         left = (uint64_t) -1;
1728
1729                 r = bus_poll(bus, true, left);
1730                 if (r < 0)
1731                         return r;
1732
1733                 r = dispatch_wqueue(bus);
1734                 if (r < 0) {
1735                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1736                                 bus_enter_closing(bus);
1737
1738                         return r;
1739                 }
1740         }
1741 }
1742
1743 _public_ int sd_bus_get_fd(sd_bus *bus) {
1744
1745         assert_return(bus, -EINVAL);
1746         assert_return(bus->input_fd == bus->output_fd, -EPERM);
1747         assert_return(!bus_pid_changed(bus), -ECHILD);
1748
1749         return bus->input_fd;
1750 }
1751
1752 _public_ int sd_bus_get_events(sd_bus *bus) {
1753         int flags = 0;
1754
1755         assert_return(bus, -EINVAL);
1756         assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1757         assert_return(!bus_pid_changed(bus), -ECHILD);
1758
1759         if (bus->state == BUS_OPENING)
1760                 flags |= POLLOUT;
1761         else if (bus->state == BUS_AUTHENTICATING) {
1762
1763                 if (bus_socket_auth_needs_write(bus))
1764                         flags |= POLLOUT;
1765
1766                 flags |= POLLIN;
1767
1768         } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
1769                 if (bus->rqueue_size <= 0)
1770                         flags |= POLLIN;
1771                 if (bus->wqueue_size > 0)
1772                         flags |= POLLOUT;
1773         }
1774
1775         return flags;
1776 }
1777
1778 _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
1779         struct reply_callback *c;
1780
1781         assert_return(bus, -EINVAL);
1782         assert_return(timeout_usec, -EINVAL);
1783         assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1784         assert_return(!bus_pid_changed(bus), -ECHILD);
1785
1786         if (bus->state == BUS_CLOSING) {
1787                 *timeout_usec = 0;
1788                 return 1;
1789         }
1790
1791         if (bus->state == BUS_AUTHENTICATING) {
1792                 *timeout_usec = bus->auth_timeout;
1793                 return 1;
1794         }
1795
1796         if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
1797                 *timeout_usec = (uint64_t) -1;
1798                 return 0;
1799         }
1800
1801         if (bus->rqueue_size > 0) {
1802                 *timeout_usec = 0;
1803                 return 1;
1804         }
1805
1806         c = prioq_peek(bus->reply_callbacks_prioq);
1807         if (!c) {
1808                 *timeout_usec = (uint64_t) -1;
1809                 return 0;
1810         }
1811
1812         *timeout_usec = c->timeout;
1813         return 1;
1814 }
1815
1816 static int process_timeout(sd_bus *bus) {
1817         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1818         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
1819         struct reply_callback *c;
1820         usec_t n;
1821         int r;
1822
1823         assert(bus);
1824
1825         c = prioq_peek(bus->reply_callbacks_prioq);
1826         if (!c)
1827                 return 0;
1828
1829         n = now(CLOCK_MONOTONIC);
1830         if (c->timeout > n)
1831                 return 0;
1832
1833         r = bus_message_new_synthetic_error(
1834                         bus,
1835                         c->serial,
1836                         &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
1837                         &m);
1838         if (r < 0)
1839                 return r;
1840
1841         m->sender = "org.freedesktop.DBus";
1842
1843         r = bus_seal_synthetic_message(bus, m);
1844         if (r < 0)
1845                 return r;
1846
1847         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
1848         hashmap_remove(bus->reply_callbacks, &c->serial);
1849
1850         bus->current = m;
1851         bus->iteration_counter ++;
1852
1853         r = c->callback(bus, m, c->userdata, &error_buffer);
1854         r = bus_maybe_reply_error(m, r, &error_buffer);
1855         free(c);
1856
1857         bus->current = NULL;
1858
1859         return r;
1860 }
1861
1862 static int process_hello(sd_bus *bus, sd_bus_message *m) {
1863         assert(bus);
1864         assert(m);
1865
1866         if (bus->state != BUS_HELLO)
1867                 return 0;
1868
1869         /* Let's make sure the first message on the bus is the HELLO
1870          * reply. But note that we don't actually parse the message
1871          * here (we leave that to the usual handling), we just verify
1872          * we don't let any earlier msg through. */
1873
1874         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1875             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1876                 return -EIO;
1877
1878         if (m->reply_serial != bus->hello_serial)
1879                 return -EIO;
1880
1881         return 0;
1882 }
1883
1884 static int process_reply(sd_bus *bus, sd_bus_message *m) {
1885         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1886         struct reply_callback *c;
1887         int r;
1888
1889         assert(bus);
1890         assert(m);
1891
1892         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1893             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1894                 return 0;
1895
1896         c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
1897         if (!c)
1898                 return 0;
1899
1900         if (c->timeout != 0)
1901                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1902
1903         r = sd_bus_message_rewind(m, true);
1904         if (r < 0)
1905                 return r;
1906
1907         r = c->callback(bus, m, c->userdata, &error_buffer);
1908         r = bus_maybe_reply_error(m, r, &error_buffer);
1909         free(c);
1910
1911         return r;
1912 }
1913
1914 static int process_filter(sd_bus *bus, sd_bus_message *m) {
1915         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1916         struct filter_callback *l;
1917         int r;
1918
1919         assert(bus);
1920         assert(m);
1921
1922         do {
1923                 bus->filter_callbacks_modified = false;
1924
1925                 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
1926
1927                         if (bus->filter_callbacks_modified)
1928                                 break;
1929
1930                         /* Don't run this more than once per iteration */
1931                         if (l->last_iteration == bus->iteration_counter)
1932                                 continue;
1933
1934                         l->last_iteration = bus->iteration_counter;
1935
1936                         r = sd_bus_message_rewind(m, true);
1937                         if (r < 0)
1938                                 return r;
1939
1940                         r = l->callback(bus, m, l->userdata, &error_buffer);
1941                         r = bus_maybe_reply_error(m, r, &error_buffer);
1942                         if (r != 0)
1943                                 return r;
1944
1945                 }
1946
1947         } while (bus->filter_callbacks_modified);
1948
1949         return 0;
1950 }
1951
1952 static int process_match(sd_bus *bus, sd_bus_message *m) {
1953         int r;
1954
1955         assert(bus);
1956         assert(m);
1957
1958         do {
1959                 bus->match_callbacks_modified = false;
1960
1961                 r = bus_match_run(bus, &bus->match_callbacks, m);
1962                 if (r != 0)
1963                         return r;
1964
1965         } while (bus->match_callbacks_modified);
1966
1967         return 0;
1968 }
1969
1970 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
1971         _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
1972         int r;
1973
1974         assert(bus);
1975         assert(m);
1976
1977         if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
1978                 return 0;
1979
1980         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
1981                 return 0;
1982
1983         if (m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED)
1984                 return 1;
1985
1986         if (streq_ptr(m->member, "Ping"))
1987                 r = sd_bus_message_new_method_return(m, &reply);
1988         else if (streq_ptr(m->member, "GetMachineId")) {
1989                 sd_id128_t id;
1990                 char sid[33];
1991
1992                 r = sd_id128_get_machine(&id);
1993                 if (r < 0)
1994                         return r;
1995
1996                 r = sd_bus_message_new_method_return(m, &reply);
1997                 if (r < 0)
1998                         return r;
1999
2000                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
2001         } else {
2002                 r = sd_bus_message_new_method_errorf(
2003                                 m, &reply,
2004                                 SD_BUS_ERROR_UNKNOWN_METHOD,
2005                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
2006         }
2007
2008         if (r < 0)
2009                 return r;
2010
2011         r = sd_bus_send(bus, reply, NULL);
2012         if (r < 0)
2013                 return r;
2014
2015         return 1;
2016 }
2017
2018 static int process_message(sd_bus *bus, sd_bus_message *m) {
2019         int r;
2020
2021         assert(bus);
2022         assert(m);
2023
2024         bus->current = m;
2025         bus->iteration_counter++;
2026
2027         log_debug("Got message sender=%s object=%s interface=%s member=%s",
2028                   strna(sd_bus_message_get_sender(m)),
2029                   strna(sd_bus_message_get_path(m)),
2030                   strna(sd_bus_message_get_interface(m)),
2031                   strna(sd_bus_message_get_member(m)));
2032
2033         r = process_hello(bus, m);
2034         if (r != 0)
2035                 goto finish;
2036
2037         r = process_reply(bus, m);
2038         if (r != 0)
2039                 goto finish;
2040
2041         r = process_filter(bus, m);
2042         if (r != 0)
2043                 goto finish;
2044
2045         r = process_match(bus, m);
2046         if (r != 0)
2047                 goto finish;
2048
2049         r = process_builtin(bus, m);
2050         if (r != 0)
2051                 goto finish;
2052
2053         r = bus_process_object(bus, m);
2054
2055 finish:
2056         bus->current = NULL;
2057         return r;
2058 }
2059
2060 static int process_running(sd_bus *bus, sd_bus_message **ret) {
2061         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2062         int r;
2063
2064         assert(bus);
2065         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
2066
2067         r = process_timeout(bus);
2068         if (r != 0)
2069                 goto null_message;
2070
2071         r = dispatch_wqueue(bus);
2072         if (r != 0)
2073                 goto null_message;
2074
2075         r = dispatch_rqueue(bus, &m);
2076         if (r < 0)
2077                 return r;
2078         if (!m)
2079                 goto null_message;
2080
2081         r = process_message(bus, m);
2082         if (r != 0)
2083                 goto null_message;
2084
2085         if (ret) {
2086                 r = sd_bus_message_rewind(m, true);
2087                 if (r < 0)
2088                         return r;
2089
2090                 *ret = m;
2091                 m = NULL;
2092                 return 1;
2093         }
2094
2095         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
2096
2097                 r = sd_bus_reply_method_errorf(
2098                                 m,
2099                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
2100                                 "Unknown object '%s'.", m->path);
2101                 if (r < 0)
2102                         return r;
2103         }
2104
2105         return 1;
2106
2107 null_message:
2108         if (r >= 0 && ret)
2109                 *ret = NULL;
2110
2111         return r;
2112 }
2113
2114 static int process_closing(sd_bus *bus, sd_bus_message **ret) {
2115         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2116         struct reply_callback *c;
2117         int r;
2118
2119         assert(bus);
2120         assert(bus->state == BUS_CLOSING);
2121
2122         c = hashmap_first(bus->reply_callbacks);
2123         if (c) {
2124                 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2125
2126                 /* First, fail all outstanding method calls */
2127                 r = bus_message_new_synthetic_error(
2128                                 bus,
2129                                 c->serial,
2130                                 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
2131                                 &m);
2132                 if (r < 0)
2133                         return r;
2134
2135                 r = bus_seal_synthetic_message(bus, m);
2136                 if (r < 0)
2137                         return r;
2138
2139                 if (c->timeout != 0)
2140                         prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
2141
2142                 hashmap_remove(bus->reply_callbacks, &c->serial);
2143
2144                 bus->current = m;
2145                 bus->iteration_counter++;
2146
2147                 r = c->callback(bus, m, c->userdata, &error_buffer);
2148                 r = bus_maybe_reply_error(m, r, &error_buffer);
2149                 free(c);
2150
2151                 goto finish;
2152         }
2153
2154         /* Then, synthesize a Disconnected message */
2155         r = sd_bus_message_new_signal(
2156                         bus,
2157                         "/org/freedesktop/DBus/Local",
2158                         "org.freedesktop.DBus.Local",
2159                         "Disconnected",
2160                         &m);
2161         if (r < 0)
2162                 return r;
2163
2164         m->sender = "org.freedesktop.DBus.Local";
2165
2166         r = bus_seal_synthetic_message(bus, m);
2167         if (r < 0)
2168                 return r;
2169
2170         sd_bus_close(bus);
2171
2172         bus->current = m;
2173         bus->iteration_counter++;
2174
2175         r = process_filter(bus, m);
2176         if (r != 0)
2177                 goto finish;
2178
2179         r = process_match(bus, m);
2180         if (r != 0)
2181                 goto finish;
2182
2183         if (ret) {
2184                 *ret = m;
2185                 m = NULL;
2186         }
2187
2188         r = 1;
2189
2190 finish:
2191         bus->current = NULL;
2192         return r;
2193 }
2194
2195 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
2196         BUS_DONT_DESTROY(bus);
2197         int r;
2198
2199         /* Returns 0 when we didn't do anything. This should cause the
2200          * caller to invoke sd_bus_wait() before returning the next
2201          * time. Returns > 0 when we did something, which possibly
2202          * means *ret is filled in with an unprocessed message. */
2203
2204         assert_return(bus, -EINVAL);
2205         assert_return(!bus_pid_changed(bus), -ECHILD);
2206
2207         /* We don't allow recursively invoking sd_bus_process(). */
2208         assert_return(!bus->current, -EBUSY);
2209
2210         switch (bus->state) {
2211
2212         case BUS_UNSET:
2213         case BUS_CLOSED:
2214                 return -ENOTCONN;
2215
2216         case BUS_OPENING:
2217                 r = bus_socket_process_opening(bus);
2218                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2219                         bus_enter_closing(bus);
2220                         r = 1;
2221                 } else if (r < 0)
2222                         return r;
2223                 if (ret)
2224                         *ret = NULL;
2225                 return r;
2226
2227         case BUS_AUTHENTICATING:
2228                 r = bus_socket_process_authenticating(bus);
2229                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2230                         bus_enter_closing(bus);
2231                         r = 1;
2232                 } else if (r < 0)
2233                         return r;
2234
2235                 if (ret)
2236                         *ret = NULL;
2237
2238                 return r;
2239
2240         case BUS_RUNNING:
2241         case BUS_HELLO:
2242                 r = process_running(bus, ret);
2243                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2244                         bus_enter_closing(bus);
2245                         r = 1;
2246
2247                         if (ret)
2248                                 *ret = NULL;
2249                 }
2250
2251                 return r;
2252
2253         case BUS_CLOSING:
2254                 return process_closing(bus, ret);
2255         }
2256
2257         assert_not_reached("Unknown state");
2258 }
2259
2260 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
2261         struct pollfd p[2] = {};
2262         int r, e, n;
2263         struct timespec ts;
2264         usec_t m = (usec_t) -1;
2265
2266         assert(bus);
2267
2268         if (bus->state == BUS_CLOSING)
2269                 return 1;
2270
2271         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2272
2273         e = sd_bus_get_events(bus);
2274         if (e < 0)
2275                 return e;
2276
2277         if (need_more)
2278                 /* The caller really needs some more data, he doesn't
2279                  * care about what's already read, or any timeouts
2280                  * except its own.*/
2281                 e |= POLLIN;
2282         else {
2283                 usec_t until;
2284                 /* The caller wants to process if there's something to
2285                  * process, but doesn't care otherwise */
2286
2287                 r = sd_bus_get_timeout(bus, &until);
2288                 if (r < 0)
2289                         return r;
2290                 if (r > 0) {
2291                         usec_t nw;
2292                         nw = now(CLOCK_MONOTONIC);
2293                         m = until > nw ? until - nw : 0;
2294                 }
2295         }
2296
2297         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
2298                 m = timeout_usec;
2299
2300         p[0].fd = bus->input_fd;
2301         if (bus->output_fd == bus->input_fd) {
2302                 p[0].events = e;
2303                 n = 1;
2304         } else {
2305                 p[0].events = e & POLLIN;
2306                 p[1].fd = bus->output_fd;
2307                 p[1].events = e & POLLOUT;
2308                 n = 2;
2309         }
2310
2311         r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
2312         if (r < 0)
2313                 return -errno;
2314
2315         return r > 0 ? 1 : 0;
2316 }
2317
2318 _public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
2319
2320         assert_return(bus, -EINVAL);
2321         assert_return(!bus_pid_changed(bus), -ECHILD);
2322
2323         if (bus->state == BUS_CLOSING)
2324                 return 0;
2325
2326         assert_return(BUS_IS_OPEN(bus->state) , -ENOTCONN);
2327
2328         if (bus->rqueue_size > 0)
2329                 return 0;
2330
2331         return bus_poll(bus, false, timeout_usec);
2332 }
2333
2334 _public_ int sd_bus_flush(sd_bus *bus) {
2335         int r;
2336
2337         assert_return(bus, -EINVAL);
2338         assert_return(!bus_pid_changed(bus), -ECHILD);
2339
2340         if (bus->state == BUS_CLOSING)
2341                 return 0;
2342
2343         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2344
2345         r = bus_ensure_running(bus);
2346         if (r < 0)
2347                 return r;
2348
2349         if (bus->wqueue_size <= 0)
2350                 return 0;
2351
2352         for (;;) {
2353                 r = dispatch_wqueue(bus);
2354                 if (r < 0) {
2355                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
2356                                 bus_enter_closing(bus);
2357
2358                         return r;
2359                 }
2360
2361                 if (bus->wqueue_size <= 0)
2362                         return 0;
2363
2364                 r = bus_poll(bus, false, (uint64_t) -1);
2365                 if (r < 0)
2366                         return r;
2367         }
2368 }
2369
2370 _public_ int sd_bus_add_filter(sd_bus *bus,
2371                                sd_bus_message_handler_t callback,
2372                                void *userdata) {
2373
2374         struct filter_callback *f;
2375
2376         assert_return(bus, -EINVAL);
2377         assert_return(callback, -EINVAL);
2378         assert_return(!bus_pid_changed(bus), -ECHILD);
2379
2380         f = new0(struct filter_callback, 1);
2381         if (!f)
2382                 return -ENOMEM;
2383         f->callback = callback;
2384         f->userdata = userdata;
2385
2386         bus->filter_callbacks_modified = true;
2387         LIST_PREPEND(callbacks, bus->filter_callbacks, f);
2388         return 0;
2389 }
2390
2391 _public_ int sd_bus_remove_filter(sd_bus *bus,
2392                                   sd_bus_message_handler_t callback,
2393                                   void *userdata) {
2394
2395         struct filter_callback *f;
2396
2397         assert_return(bus, -EINVAL);
2398         assert_return(callback, -EINVAL);
2399         assert_return(!bus_pid_changed(bus), -ECHILD);
2400
2401         LIST_FOREACH(callbacks, f, bus->filter_callbacks) {
2402                 if (f->callback == callback && f->userdata == userdata) {
2403                         bus->filter_callbacks_modified = true;
2404                         LIST_REMOVE(callbacks, bus->filter_callbacks, f);
2405                         free(f);
2406                         return 1;
2407                 }
2408         }
2409
2410         return 0;
2411 }
2412
2413 _public_ int sd_bus_add_match(sd_bus *bus,
2414                               const char *match,
2415                               sd_bus_message_handler_t callback,
2416                               void *userdata) {
2417
2418         struct bus_match_component *components = NULL;
2419         unsigned n_components = 0;
2420         uint64_t cookie = 0;
2421         int r = 0;
2422
2423         assert_return(bus, -EINVAL);
2424         assert_return(match, -EINVAL);
2425         assert_return(!bus_pid_changed(bus), -ECHILD);
2426
2427         r = bus_match_parse(match, &components, &n_components);
2428         if (r < 0)
2429                 goto finish;
2430
2431         if (bus->bus_client) {
2432                 cookie = ++bus->match_cookie;
2433
2434                 r = bus_add_match_internal(bus, match, components, n_components, cookie);
2435                 if (r < 0)
2436                         goto finish;
2437         }
2438
2439         bus->match_callbacks_modified = true;
2440         r = bus_match_add(&bus->match_callbacks, components, n_components, callback, userdata, cookie, NULL);
2441         if (r < 0) {
2442                 if (bus->bus_client)
2443                         bus_remove_match_internal(bus, match, cookie);
2444         }
2445
2446 finish:
2447         bus_match_parse_free(components, n_components);
2448         return r;
2449 }
2450
2451 _public_ int sd_bus_remove_match(sd_bus *bus,
2452                                  const char *match,
2453                                  sd_bus_message_handler_t callback,
2454                                  void *userdata) {
2455
2456         struct bus_match_component *components = NULL;
2457         unsigned n_components = 0;
2458         int r = 0, q = 0;
2459         uint64_t cookie = 0;
2460
2461         assert_return(bus, -EINVAL);
2462         assert_return(match, -EINVAL);
2463         assert_return(!bus_pid_changed(bus), -ECHILD);
2464
2465         r = bus_match_parse(match, &components, &n_components);
2466         if (r < 0)
2467                 return r;
2468
2469         bus->match_callbacks_modified = true;
2470         r = bus_match_remove(&bus->match_callbacks, components, n_components, callback, userdata, &cookie);
2471
2472         if (bus->bus_client)
2473                 q = bus_remove_match_internal(bus, match, cookie);
2474
2475         bus_match_parse_free(components, n_components);
2476
2477         return r < 0 ? r : q;
2478 }
2479
2480 bool bus_pid_changed(sd_bus *bus) {
2481         assert(bus);
2482
2483         /* We don't support people creating a bus connection and
2484          * keeping it around over a fork(). Let's complain. */
2485
2486         return bus->original_pid != getpid();
2487 }
2488
2489 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
2490         sd_bus *bus = userdata;
2491         int r;
2492
2493         assert(bus);
2494
2495         r = sd_bus_process(bus, NULL);
2496         if (r < 0)
2497                 return r;
2498
2499         return 1;
2500 }
2501
2502 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
2503         sd_bus *bus = userdata;
2504         int r;
2505
2506         assert(bus);
2507
2508         r = sd_bus_process(bus, NULL);
2509         if (r < 0)
2510                 return r;
2511
2512         return 1;
2513 }
2514
2515 static int prepare_callback(sd_event_source *s, void *userdata) {
2516         sd_bus *bus = userdata;
2517         int r, e;
2518         usec_t until;
2519
2520         assert(s);
2521         assert(bus);
2522
2523         e = sd_bus_get_events(bus);
2524         if (e < 0)
2525                 return e;
2526
2527         if (bus->output_fd != bus->input_fd) {
2528
2529                 r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
2530                 if (r < 0)
2531                         return r;
2532
2533                 r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
2534                 if (r < 0)
2535                         return r;
2536         } else {
2537                 r = sd_event_source_set_io_events(bus->input_io_event_source, e);
2538                 if (r < 0)
2539                         return r;
2540         }
2541
2542         r = sd_bus_get_timeout(bus, &until);
2543         if (r < 0)
2544                 return r;
2545         if (r > 0) {
2546                 int j;
2547
2548                 j = sd_event_source_set_time(bus->time_event_source, until);
2549                 if (j < 0)
2550                         return j;
2551         }
2552
2553         r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
2554         if (r < 0)
2555                 return r;
2556
2557         return 1;
2558 }
2559
2560 static int quit_callback(sd_event_source *event, void *userdata) {
2561         sd_bus *bus = userdata;
2562
2563         assert(event);
2564
2565         sd_bus_flush(bus);
2566
2567         return 1;
2568 }
2569
2570 _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
2571         int r;
2572
2573         assert_return(bus, -EINVAL);
2574         assert_return(!bus->event, -EBUSY);
2575
2576         assert(!bus->input_io_event_source);
2577         assert(!bus->output_io_event_source);
2578         assert(!bus->time_event_source);
2579
2580         if (event)
2581                 bus->event = sd_event_ref(event);
2582         else  {
2583                 r = sd_event_default(&bus->event);
2584                 if (r < 0)
2585                         return r;
2586         }
2587
2588         r = sd_event_add_io(bus->event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
2589         if (r < 0)
2590                 goto fail;
2591
2592         r = sd_event_source_set_priority(bus->input_io_event_source, priority);
2593         if (r < 0)
2594                 goto fail;
2595
2596         if (bus->output_fd != bus->input_fd) {
2597                 r = sd_event_add_io(bus->event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
2598                 if (r < 0)
2599                         goto fail;
2600
2601                 r = sd_event_source_set_priority(bus->output_io_event_source, priority);
2602                 if (r < 0)
2603                         goto fail;
2604         }
2605
2606         r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
2607         if (r < 0)
2608                 goto fail;
2609
2610         r = sd_event_add_monotonic(bus->event, 0, 0, time_callback, bus, &bus->time_event_source);
2611         if (r < 0)
2612                 goto fail;
2613
2614         r = sd_event_source_set_priority(bus->time_event_source, priority);
2615         if (r < 0)
2616                 goto fail;
2617
2618         r = sd_event_add_quit(bus->event, quit_callback, bus, &bus->quit_event_source);
2619         if (r < 0)
2620                 goto fail;
2621
2622         return 0;
2623
2624 fail:
2625         sd_bus_detach_event(bus);
2626         return r;
2627 }
2628
2629 _public_ int sd_bus_detach_event(sd_bus *bus) {
2630         assert_return(bus, -EINVAL);
2631         assert_return(bus->event, -ENXIO);
2632
2633         if (bus->input_io_event_source) {
2634                 sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
2635                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
2636         }
2637
2638         if (bus->output_io_event_source) {
2639                 sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
2640                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
2641         }
2642
2643         if (bus->time_event_source) {
2644                 sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
2645                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
2646         }
2647
2648         if (bus->quit_event_source) {
2649                 sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
2650                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
2651         }
2652
2653         if (bus->event)
2654                 bus->event = sd_event_unref(bus->event);
2655
2656         return 0;
2657 }
2658
2659 _public_ sd_event* sd_bus_get_event(sd_bus *bus) {
2660         assert_return(bus, NULL);
2661
2662         return bus->event;
2663 }
2664
2665 _public_ sd_bus_message* sd_bus_get_current(sd_bus *bus) {
2666         assert_return(bus, NULL);
2667
2668         return bus->current;
2669 }
2670
2671 static int bus_default(int (*bus_open)(sd_bus **), sd_bus **default_bus, sd_bus **ret) {
2672         sd_bus *b = NULL;
2673         int r;
2674
2675         assert(bus_open);
2676         assert(default_bus);
2677
2678         if (!ret)
2679                 return !!*default_bus;
2680
2681         if (*default_bus) {
2682                 *ret = sd_bus_ref(*default_bus);
2683                 return 0;
2684         }
2685
2686         r = bus_open(&b);
2687         if (r < 0)
2688                 return r;
2689
2690         b->default_bus_ptr = default_bus;
2691         b->tid = gettid();
2692         *default_bus = b;
2693
2694         *ret = b;
2695         return 1;
2696 }
2697
2698 _public_ int sd_bus_default_system(sd_bus **ret) {
2699         static __thread sd_bus *default_system_bus = NULL;
2700
2701         return bus_default(sd_bus_open_system, &default_system_bus, ret);
2702 }
2703
2704 _public_ int sd_bus_default_user(sd_bus **ret) {
2705         static __thread sd_bus *default_user_bus = NULL;
2706
2707         return bus_default(sd_bus_open_user, &default_user_bus, ret);
2708 }
2709
2710 _public_ int sd_bus_get_tid(sd_bus *b, pid_t *tid) {
2711         assert_return(b, -EINVAL);
2712         assert_return(tid, -EINVAL);
2713         assert_return(!bus_pid_changed(b), -ECHILD);
2714
2715         if (b->tid != 0) {
2716                 *tid = b->tid;
2717                 return 0;
2718         }
2719
2720         if (b->event)
2721                 return sd_event_get_tid(b->event, tid);
2722
2723         return -ENXIO;
2724 }
2725
2726 _public_ char *sd_bus_label_escape(const char *s) {
2727         char *r, *t;
2728         const char *f;
2729
2730         assert_return(s, NULL);
2731
2732         /* Escapes all chars that D-Bus' object path cannot deal
2733          * with. Can be reversed with bus_path_unescape(). We special
2734          * case the empty string. */
2735
2736         if (*s == 0)
2737                 return strdup("_");
2738
2739         r = new(char, strlen(s)*3 + 1);
2740         if (!r)
2741                 return NULL;
2742
2743         for (f = s, t = r; *f; f++) {
2744
2745                 /* Escape everything that is not a-zA-Z0-9. We also
2746                  * escape 0-9 if it's the first character */
2747
2748                 if (!(*f >= 'A' && *f <= 'Z') &&
2749                     !(*f >= 'a' && *f <= 'z') &&
2750                     !(f > s && *f >= '0' && *f <= '9')) {
2751                         *(t++) = '_';
2752                         *(t++) = hexchar(*f >> 4);
2753                         *(t++) = hexchar(*f);
2754                 } else
2755                         *(t++) = *f;
2756         }
2757
2758         *t = 0;
2759
2760         return r;
2761 }
2762
2763 _public_ char *sd_bus_label_unescape(const char *f) {
2764         char *r, *t;
2765
2766         assert_return(f, NULL);
2767
2768         /* Special case for the empty string */
2769         if (streq(f, "_"))
2770                 return strdup("");
2771
2772         r = new(char, strlen(f) + 1);
2773         if (!r)
2774                 return NULL;
2775
2776         for (t = r; *f; f++) {
2777
2778                 if (*f == '_') {
2779                         int a, b;
2780
2781                         if ((a = unhexchar(f[1])) < 0 ||
2782                             (b = unhexchar(f[2])) < 0) {
2783                                 /* Invalid escape code, let's take it literal then */
2784                                 *(t++) = '_';
2785                         } else {
2786                                 *(t++) = (char) ((a << 4) | b);
2787                                 f += 2;
2788                         }
2789                 } else
2790                         *(t++) = *f;
2791         }
2792
2793         *t = 0;
2794
2795         return r;
2796 }
2797
2798 _public_ int sd_bus_get_peer_creds(sd_bus *bus, uint64_t mask, sd_bus_creds **ret) {
2799         sd_bus_creds *c;
2800         pid_t pid = 0;
2801         int r;
2802
2803         assert_return(bus, -EINVAL);
2804         assert_return(mask <= _SD_BUS_CREDS_MAX, -ENOTSUP);
2805         assert_return(ret, -EINVAL);
2806         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2807         assert_return(!bus_pid_changed(bus), -ECHILD);
2808         assert_return(!bus->is_kernel, -ENOTSUP);
2809
2810         if (!bus->ucred_valid && !isempty(bus->label))
2811                 return -ENODATA;
2812
2813         c = bus_creds_new();
2814         if (!c)
2815                 return -ENOMEM;
2816
2817         if (bus->ucred_valid) {
2818                 pid = c->pid = bus->ucred.pid;
2819                 c->uid = bus->ucred.uid;
2820                 c->gid = bus->ucred.gid;
2821
2822                 c->mask |= (SD_BUS_CREDS_UID | SD_BUS_CREDS_PID | SD_BUS_CREDS_GID) & mask;
2823         }
2824
2825         if (!isempty(bus->label) && (mask & SD_BUS_CREDS_SELINUX_CONTEXT)) {
2826                 c->label = strdup(bus->label);
2827                 if (!c->label) {
2828                         sd_bus_creds_unref(c);
2829                         return -ENOMEM;
2830                 }
2831
2832                 c->mask |= SD_BUS_CREDS_SELINUX_CONTEXT;
2833         }
2834
2835         r = bus_creds_add_more(c, mask, pid, 0);
2836         if (r < 0)
2837                 return r;
2838
2839         *ret = c;
2840         return 0;
2841 }