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_ALL, -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         assert(bus);
1288         assert(message);
1289
1290         if (bus->is_kernel)
1291                 return bus_kernel_write_message(bus, message);
1292         else
1293                 return bus_socket_write_message(bus, message, idx);
1294 }
1295
1296 static int dispatch_wqueue(sd_bus *bus) {
1297         int r, ret = 0;
1298
1299         assert(bus);
1300         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1301
1302         while (bus->wqueue_size > 0) {
1303
1304                 r = bus_write_message(bus, bus->wqueue[0], &bus->windex);
1305                 if (r < 0)
1306                         return r;
1307                 else if (r == 0)
1308                         /* Didn't do anything this time */
1309                         return ret;
1310                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1311                         /* Fully written. Let's drop the entry from
1312                          * the queue.
1313                          *
1314                          * This isn't particularly optimized, but
1315                          * well, this is supposed to be our worst-case
1316                          * buffer only, and the socket buffer is
1317                          * supposed to be our primary buffer, and if
1318                          * it got full, then all bets are off
1319                          * anyway. */
1320
1321                         sd_bus_message_unref(bus->wqueue[0]);
1322                         bus->wqueue_size --;
1323                         memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1324                         bus->windex = 0;
1325
1326                         ret = 1;
1327                 }
1328         }
1329
1330         return ret;
1331 }
1332
1333 static int bus_read_message(sd_bus *bus) {
1334         assert(bus);
1335
1336         if (bus->is_kernel)
1337                 return bus_kernel_read_message(bus);
1338         else
1339                 return bus_socket_read_message(bus);
1340 }
1341
1342 int bus_rqueue_make_room(sd_bus *bus) {
1343         sd_bus_message **q;
1344         unsigned x;
1345
1346         x = bus->rqueue_size + 1;
1347
1348         if (bus->rqueue_allocated >= x)
1349                 return 0;
1350
1351         if (x > BUS_RQUEUE_MAX)
1352                 return -ENOBUFS;
1353
1354         q = realloc(bus->rqueue, x * sizeof(sd_bus_message*));
1355         if (!q)
1356                 return -ENOMEM;
1357
1358         bus->rqueue = q;
1359         bus->rqueue_allocated = x;
1360
1361         return 0;
1362 }
1363
1364 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
1365         int r, ret = 0;
1366
1367         assert(bus);
1368         assert(m);
1369         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1370
1371         for (;;) {
1372                 if (bus->rqueue_size > 0) {
1373                         /* Dispatch a queued message */
1374
1375                         *m = bus->rqueue[0];
1376                         bus->rqueue_size --;
1377                         memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1378                         return 1;
1379                 }
1380
1381                 /* Try to read a new message */
1382                 r = bus_read_message(bus);
1383                 if (r < 0)
1384                         return r;
1385                 if (r == 0)
1386                         return ret;
1387
1388                 ret = 1;
1389         }
1390 }
1391
1392 _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
1393         int r;
1394
1395         assert_return(bus, -EINVAL);
1396         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1397         assert_return(m, -EINVAL);
1398         assert_return(!bus_pid_changed(bus), -ECHILD);
1399
1400         if (m->n_fds > 0) {
1401                 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1402                 if (r < 0)
1403                         return r;
1404                 if (r == 0)
1405                         return -ENOTSUP;
1406         }
1407
1408         /* If the serial number isn't kept, then we know that no reply
1409          * is expected */
1410         if (!serial && !m->sealed)
1411                 m->header->flags |= SD_BUS_MESSAGE_NO_REPLY_EXPECTED;
1412
1413         r = bus_seal_message(bus, m);
1414         if (r < 0)
1415                 return r;
1416
1417         /* If this is a reply and no reply was requested, then let's
1418          * suppress this, if we can */
1419         if (m->dont_send && !serial)
1420                 return 1;
1421
1422         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1423                 size_t idx = 0;
1424
1425                 r = bus_write_message(bus, m, &idx);
1426                 if (r < 0) {
1427                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1428                                 bus_enter_closing(bus);
1429
1430                         return r;
1431                 } else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
1432                         /* Wasn't fully written. So let's remember how
1433                          * much was written. Note that the first entry
1434                          * of the wqueue array is always allocated so
1435                          * that we always can remember how much was
1436                          * written. */
1437                         bus->wqueue[0] = sd_bus_message_ref(m);
1438                         bus->wqueue_size = 1;
1439                         bus->windex = idx;
1440                 }
1441         } else {
1442                 sd_bus_message **q;
1443
1444                 /* Just append it to the queue. */
1445
1446                 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1447                         return -ENOBUFS;
1448
1449                 q = realloc(bus->wqueue, sizeof(sd_bus_message*) * (bus->wqueue_size + 1));
1450                 if (!q)
1451                         return -ENOMEM;
1452
1453                 bus->wqueue = q;
1454                 q[bus->wqueue_size ++] = sd_bus_message_ref(m);
1455         }
1456
1457         if (serial)
1458                 *serial = BUS_MESSAGE_SERIAL(m);
1459
1460         return 1;
1461 }
1462
1463 _public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *serial) {
1464         int r;
1465
1466         assert_return(bus, -EINVAL);
1467         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1468         assert_return(m, -EINVAL);
1469         assert_return(!bus_pid_changed(bus), -ECHILD);
1470
1471         if (!streq_ptr(m->destination, destination)) {
1472
1473                 if (!destination)
1474                         return -EEXIST;
1475
1476                 r = sd_bus_message_set_destination(m, destination);
1477                 if (r < 0)
1478                         return r;
1479         }
1480
1481         return sd_bus_send(bus, m, serial);
1482 }
1483
1484 static usec_t calc_elapse(uint64_t usec) {
1485         if (usec == (uint64_t) -1)
1486                 return 0;
1487
1488         if (usec == 0)
1489                 usec = BUS_DEFAULT_TIMEOUT;
1490
1491         return now(CLOCK_MONOTONIC) + usec;
1492 }
1493
1494 static int timeout_compare(const void *a, const void *b) {
1495         const struct reply_callback *x = a, *y = b;
1496
1497         if (x->timeout != 0 && y->timeout == 0)
1498                 return -1;
1499
1500         if (x->timeout == 0 && y->timeout != 0)
1501                 return 1;
1502
1503         if (x->timeout < y->timeout)
1504                 return -1;
1505
1506         if (x->timeout > y->timeout)
1507                 return 1;
1508
1509         return 0;
1510 }
1511
1512 _public_ int sd_bus_call_async(
1513                 sd_bus *bus,
1514                 sd_bus_message *m,
1515                 sd_bus_message_handler_t callback,
1516                 void *userdata,
1517                 uint64_t usec,
1518                 uint64_t *serial) {
1519
1520         struct reply_callback *c;
1521         int r;
1522
1523         assert_return(bus, -EINVAL);
1524         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1525         assert_return(m, -EINVAL);
1526         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1527         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1528         assert_return(callback, -EINVAL);
1529         assert_return(!bus_pid_changed(bus), -ECHILD);
1530
1531         r = hashmap_ensure_allocated(&bus->reply_callbacks, uint64_hash_func, uint64_compare_func);
1532         if (r < 0)
1533                 return r;
1534
1535         if (usec != (uint64_t) -1) {
1536                 r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1537                 if (r < 0)
1538                         return r;
1539         }
1540
1541         r = bus_seal_message(bus, m);
1542         if (r < 0)
1543                 return r;
1544
1545         c = new0(struct reply_callback, 1);
1546         if (!c)
1547                 return -ENOMEM;
1548
1549         c->callback = callback;
1550         c->userdata = userdata;
1551         c->serial = BUS_MESSAGE_SERIAL(m);
1552         c->timeout = calc_elapse(usec);
1553
1554         r = hashmap_put(bus->reply_callbacks, &c->serial, c);
1555         if (r < 0) {
1556                 free(c);
1557                 return r;
1558         }
1559
1560         if (c->timeout != 0) {
1561                 r = prioq_put(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1562                 if (r < 0) {
1563                         c->timeout = 0;
1564                         sd_bus_call_async_cancel(bus, c->serial);
1565                         return r;
1566                 }
1567         }
1568
1569         r = sd_bus_send(bus, m, serial);
1570         if (r < 0) {
1571                 sd_bus_call_async_cancel(bus, c->serial);
1572                 return r;
1573         }
1574
1575         return r;
1576 }
1577
1578 _public_ int sd_bus_call_async_cancel(sd_bus *bus, uint64_t serial) {
1579         struct reply_callback *c;
1580
1581         assert_return(bus, -EINVAL);
1582         assert_return(serial != 0, -EINVAL);
1583         assert_return(!bus_pid_changed(bus), -ECHILD);
1584
1585         c = hashmap_remove(bus->reply_callbacks, &serial);
1586         if (!c)
1587                 return 0;
1588
1589         if (c->timeout != 0)
1590                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1591
1592         free(c);
1593         return 1;
1594 }
1595
1596 int bus_ensure_running(sd_bus *bus) {
1597         int r;
1598
1599         assert(bus);
1600
1601         if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
1602                 return -ENOTCONN;
1603         if (bus->state == BUS_RUNNING)
1604                 return 1;
1605
1606         for (;;) {
1607                 r = sd_bus_process(bus, NULL);
1608                 if (r < 0)
1609                         return r;
1610                 if (bus->state == BUS_RUNNING)
1611                         return 1;
1612                 if (r > 0)
1613                         continue;
1614
1615                 r = sd_bus_wait(bus, (uint64_t) -1);
1616                 if (r < 0)
1617                         return r;
1618         }
1619 }
1620
1621 _public_ int sd_bus_call(
1622                 sd_bus *bus,
1623                 sd_bus_message *m,
1624                 uint64_t usec,
1625                 sd_bus_error *error,
1626                 sd_bus_message **reply) {
1627
1628         usec_t timeout;
1629         uint64_t serial;
1630         unsigned i;
1631         int r;
1632
1633         assert_return(bus, -EINVAL);
1634         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1635         assert_return(m, -EINVAL);
1636         assert_return(m->header->type == SD_BUS_MESSAGE_METHOD_CALL, -EINVAL);
1637         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1638         assert_return(!bus_error_is_dirty(error), -EINVAL);
1639         assert_return(!bus_pid_changed(bus), -ECHILD);
1640
1641         r = bus_ensure_running(bus);
1642         if (r < 0)
1643                 return r;
1644
1645         i = bus->rqueue_size;
1646
1647         r = sd_bus_send(bus, m, &serial);
1648         if (r < 0)
1649                 return r;
1650
1651         timeout = calc_elapse(usec);
1652
1653         for (;;) {
1654                 usec_t left;
1655
1656                 while (i < bus->rqueue_size) {
1657                         sd_bus_message *incoming = NULL;
1658
1659                         incoming = bus->rqueue[i];
1660
1661                         if (incoming->reply_serial == serial) {
1662                                 /* Found a match! */
1663
1664                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
1665                                 bus->rqueue_size--;
1666
1667                                 if (incoming->header->type == SD_BUS_MESSAGE_METHOD_RETURN) {
1668
1669                                         if (reply)
1670                                                 *reply = incoming;
1671                                         else
1672                                                 sd_bus_message_unref(incoming);
1673
1674                                         return 1;
1675                                 } else if (incoming->header->type == SD_BUS_MESSAGE_METHOD_ERROR)
1676                                         r = sd_bus_error_copy(error, &incoming->error);
1677                                 else
1678                                         r = -EIO;
1679
1680                                 sd_bus_message_unref(incoming);
1681                                 return r;
1682
1683                         } else if (incoming->header->serial == serial &&
1684                                    bus->unique_name &&
1685                                    incoming->sender &&
1686                                    streq(bus->unique_name, incoming->sender)) {
1687
1688                                 memmove(bus->rqueue + i, bus->rqueue + i + 1, sizeof(sd_bus_message*) * (bus->rqueue_size - i - 1));
1689                                 bus->rqueue_size--;
1690
1691                                 /* Our own message? Somebody is trying
1692                                  * to send its own client a message,
1693                                  * let's not dead-lock, let's fail
1694                                  * immediately. */
1695
1696                                 sd_bus_message_unref(incoming);
1697                                 return -ELOOP;
1698                         }
1699
1700                         /* Try to read more, right-away */
1701                         i++;
1702                 }
1703
1704                 r = bus_read_message(bus);
1705                 if (r < 0) {
1706                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1707                                 bus_enter_closing(bus);
1708
1709                         return r;
1710                 }
1711                 if (r > 0)
1712                         continue;
1713
1714                 if (timeout > 0) {
1715                         usec_t n;
1716
1717                         n = now(CLOCK_MONOTONIC);
1718                         if (n >= timeout)
1719                                 return -ETIMEDOUT;
1720
1721                         left = timeout - n;
1722                 } else
1723                         left = (uint64_t) -1;
1724
1725                 r = bus_poll(bus, true, left);
1726                 if (r < 0)
1727                         return r;
1728
1729                 r = dispatch_wqueue(bus);
1730                 if (r < 0) {
1731                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
1732                                 bus_enter_closing(bus);
1733
1734                         return r;
1735                 }
1736         }
1737 }
1738
1739 _public_ int sd_bus_get_fd(sd_bus *bus) {
1740
1741         assert_return(bus, -EINVAL);
1742         assert_return(bus->input_fd == bus->output_fd, -EPERM);
1743         assert_return(!bus_pid_changed(bus), -ECHILD);
1744
1745         return bus->input_fd;
1746 }
1747
1748 _public_ int sd_bus_get_events(sd_bus *bus) {
1749         int flags = 0;
1750
1751         assert_return(bus, -EINVAL);
1752         assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1753         assert_return(!bus_pid_changed(bus), -ECHILD);
1754
1755         if (bus->state == BUS_OPENING)
1756                 flags |= POLLOUT;
1757         else if (bus->state == BUS_AUTHENTICATING) {
1758
1759                 if (bus_socket_auth_needs_write(bus))
1760                         flags |= POLLOUT;
1761
1762                 flags |= POLLIN;
1763
1764         } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
1765                 if (bus->rqueue_size <= 0)
1766                         flags |= POLLIN;
1767                 if (bus->wqueue_size > 0)
1768                         flags |= POLLOUT;
1769         }
1770
1771         return flags;
1772 }
1773
1774 _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
1775         struct reply_callback *c;
1776
1777         assert_return(bus, -EINVAL);
1778         assert_return(timeout_usec, -EINVAL);
1779         assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
1780         assert_return(!bus_pid_changed(bus), -ECHILD);
1781
1782         if (bus->state == BUS_CLOSING) {
1783                 *timeout_usec = 0;
1784                 return 1;
1785         }
1786
1787         if (bus->state == BUS_AUTHENTICATING) {
1788                 *timeout_usec = bus->auth_timeout;
1789                 return 1;
1790         }
1791
1792         if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
1793                 *timeout_usec = (uint64_t) -1;
1794                 return 0;
1795         }
1796
1797         if (bus->rqueue_size > 0) {
1798                 *timeout_usec = 0;
1799                 return 1;
1800         }
1801
1802         c = prioq_peek(bus->reply_callbacks_prioq);
1803         if (!c) {
1804                 *timeout_usec = (uint64_t) -1;
1805                 return 0;
1806         }
1807
1808         *timeout_usec = c->timeout;
1809         return 1;
1810 }
1811
1812 static int process_timeout(sd_bus *bus) {
1813         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1814         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
1815         struct reply_callback *c;
1816         usec_t n;
1817         int r;
1818
1819         assert(bus);
1820
1821         c = prioq_peek(bus->reply_callbacks_prioq);
1822         if (!c)
1823                 return 0;
1824
1825         n = now(CLOCK_MONOTONIC);
1826         if (c->timeout > n)
1827                 return 0;
1828
1829         r = bus_message_new_synthetic_error(
1830                         bus,
1831                         c->serial,
1832                         &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
1833                         &m);
1834         if (r < 0)
1835                 return r;
1836
1837         m->sender = "org.freedesktop.DBus";
1838
1839         r = bus_seal_synthetic_message(bus, m);
1840         if (r < 0)
1841                 return r;
1842
1843         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
1844         hashmap_remove(bus->reply_callbacks, &c->serial);
1845
1846         bus->current = m;
1847         bus->iteration_counter ++;
1848
1849         r = c->callback(bus, m, c->userdata, &error_buffer);
1850         r = bus_maybe_reply_error(m, r, &error_buffer);
1851         free(c);
1852
1853         bus->current = NULL;
1854
1855         return r;
1856 }
1857
1858 static int process_hello(sd_bus *bus, sd_bus_message *m) {
1859         assert(bus);
1860         assert(m);
1861
1862         if (bus->state != BUS_HELLO)
1863                 return 0;
1864
1865         /* Let's make sure the first message on the bus is the HELLO
1866          * reply. But note that we don't actually parse the message
1867          * here (we leave that to the usual handling), we just verify
1868          * we don't let any earlier msg through. */
1869
1870         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1871             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1872                 return -EIO;
1873
1874         if (m->reply_serial != bus->hello_serial)
1875                 return -EIO;
1876
1877         return 0;
1878 }
1879
1880 static int process_reply(sd_bus *bus, sd_bus_message *m) {
1881         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1882         struct reply_callback *c;
1883         int r;
1884
1885         assert(bus);
1886         assert(m);
1887
1888         if (m->header->type != SD_BUS_MESSAGE_METHOD_RETURN &&
1889             m->header->type != SD_BUS_MESSAGE_METHOD_ERROR)
1890                 return 0;
1891
1892         c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
1893         if (!c)
1894                 return 0;
1895
1896         if (c->timeout != 0)
1897                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1898
1899         r = sd_bus_message_rewind(m, true);
1900         if (r < 0)
1901                 return r;
1902
1903         r = c->callback(bus, m, c->userdata, &error_buffer);
1904         r = bus_maybe_reply_error(m, r, &error_buffer);
1905         free(c);
1906
1907         return r;
1908 }
1909
1910 static int process_filter(sd_bus *bus, sd_bus_message *m) {
1911         _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
1912         struct filter_callback *l;
1913         int r;
1914
1915         assert(bus);
1916         assert(m);
1917
1918         do {
1919                 bus->filter_callbacks_modified = false;
1920
1921                 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
1922
1923                         if (bus->filter_callbacks_modified)
1924                                 break;
1925
1926                         /* Don't run this more than once per iteration */
1927                         if (l->last_iteration == bus->iteration_counter)
1928                                 continue;
1929
1930                         l->last_iteration = bus->iteration_counter;
1931
1932                         r = sd_bus_message_rewind(m, true);
1933                         if (r < 0)
1934                                 return r;
1935
1936                         r = l->callback(bus, m, l->userdata, &error_buffer);
1937                         r = bus_maybe_reply_error(m, r, &error_buffer);
1938                         if (r != 0)
1939                                 return r;
1940
1941                 }
1942
1943         } while (bus->filter_callbacks_modified);
1944
1945         return 0;
1946 }
1947
1948 static int process_match(sd_bus *bus, sd_bus_message *m) {
1949         int r;
1950
1951         assert(bus);
1952         assert(m);
1953
1954         do {
1955                 bus->match_callbacks_modified = false;
1956
1957                 r = bus_match_run(bus, &bus->match_callbacks, m);
1958                 if (r != 0)
1959                         return r;
1960
1961         } while (bus->match_callbacks_modified);
1962
1963         return 0;
1964 }
1965
1966 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
1967         _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
1968         int r;
1969
1970         assert(bus);
1971         assert(m);
1972
1973         if (m->header->type != SD_BUS_MESSAGE_METHOD_CALL)
1974                 return 0;
1975
1976         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
1977                 return 0;
1978
1979         if (m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED)
1980                 return 1;
1981
1982         if (streq_ptr(m->member, "Ping"))
1983                 r = sd_bus_message_new_method_return(m, &reply);
1984         else if (streq_ptr(m->member, "GetMachineId")) {
1985                 sd_id128_t id;
1986                 char sid[33];
1987
1988                 r = sd_id128_get_machine(&id);
1989                 if (r < 0)
1990                         return r;
1991
1992                 r = sd_bus_message_new_method_return(m, &reply);
1993                 if (r < 0)
1994                         return r;
1995
1996                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
1997         } else {
1998                 r = sd_bus_message_new_method_errorf(
1999                                 m, &reply,
2000                                 SD_BUS_ERROR_UNKNOWN_METHOD,
2001                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
2002         }
2003
2004         if (r < 0)
2005                 return r;
2006
2007         r = sd_bus_send(bus, reply, NULL);
2008         if (r < 0)
2009                 return r;
2010
2011         return 1;
2012 }
2013
2014 static int process_message(sd_bus *bus, sd_bus_message *m) {
2015         int r;
2016
2017         assert(bus);
2018         assert(m);
2019
2020         bus->current = m;
2021         bus->iteration_counter++;
2022
2023         log_debug("Got message sender=%s object=%s interface=%s member=%s",
2024                   strna(sd_bus_message_get_sender(m)),
2025                   strna(sd_bus_message_get_path(m)),
2026                   strna(sd_bus_message_get_interface(m)),
2027                   strna(sd_bus_message_get_member(m)));
2028
2029         r = process_hello(bus, m);
2030         if (r != 0)
2031                 goto finish;
2032
2033         r = process_reply(bus, m);
2034         if (r != 0)
2035                 goto finish;
2036
2037         r = process_filter(bus, m);
2038         if (r != 0)
2039                 goto finish;
2040
2041         r = process_match(bus, m);
2042         if (r != 0)
2043                 goto finish;
2044
2045         r = process_builtin(bus, m);
2046         if (r != 0)
2047                 goto finish;
2048
2049         r = bus_process_object(bus, m);
2050
2051 finish:
2052         bus->current = NULL;
2053         return r;
2054 }
2055
2056 static int process_running(sd_bus *bus, sd_bus_message **ret) {
2057         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2058         int r;
2059
2060         assert(bus);
2061         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
2062
2063         r = process_timeout(bus);
2064         if (r != 0)
2065                 goto null_message;
2066
2067         r = dispatch_wqueue(bus);
2068         if (r != 0)
2069                 goto null_message;
2070
2071         r = dispatch_rqueue(bus, &m);
2072         if (r < 0)
2073                 return r;
2074         if (!m)
2075                 goto null_message;
2076
2077         r = process_message(bus, m);
2078         if (r != 0)
2079                 goto null_message;
2080
2081         if (ret) {
2082                 r = sd_bus_message_rewind(m, true);
2083                 if (r < 0)
2084                         return r;
2085
2086                 *ret = m;
2087                 m = NULL;
2088                 return 1;
2089         }
2090
2091         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
2092
2093                 r = sd_bus_reply_method_errorf(
2094                                 m,
2095                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
2096                                 "Unknown object '%s'.", m->path);
2097                 if (r < 0)
2098                         return r;
2099         }
2100
2101         return 1;
2102
2103 null_message:
2104         if (r >= 0 && ret)
2105                 *ret = NULL;
2106
2107         return r;
2108 }
2109
2110 static int process_closing(sd_bus *bus, sd_bus_message **ret) {
2111         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
2112         struct reply_callback *c;
2113         int r;
2114
2115         assert(bus);
2116         assert(bus->state == BUS_CLOSING);
2117
2118         c = hashmap_first(bus->reply_callbacks);
2119         if (c) {
2120                 _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
2121
2122                 /* First, fail all outstanding method calls */
2123                 r = bus_message_new_synthetic_error(
2124                                 bus,
2125                                 c->serial,
2126                                 &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
2127                                 &m);
2128                 if (r < 0)
2129                         return r;
2130
2131                 r = bus_seal_synthetic_message(bus, m);
2132                 if (r < 0)
2133                         return r;
2134
2135                 if (c->timeout != 0)
2136                         prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
2137
2138                 hashmap_remove(bus->reply_callbacks, &c->serial);
2139
2140                 bus->current = m;
2141                 bus->iteration_counter++;
2142
2143                 r = c->callback(bus, m, c->userdata, &error_buffer);
2144                 r = bus_maybe_reply_error(m, r, &error_buffer);
2145                 free(c);
2146
2147                 goto finish;
2148         }
2149
2150         /* Then, synthesize a Disconnected message */
2151         r = sd_bus_message_new_signal(
2152                         bus,
2153                         "/org/freedesktop/DBus/Local",
2154                         "org.freedesktop.DBus.Local",
2155                         "Disconnected",
2156                         &m);
2157         if (r < 0)
2158                 return r;
2159
2160         m->sender = "org.freedesktop.DBus.Local";
2161
2162         r = bus_seal_synthetic_message(bus, m);
2163         if (r < 0)
2164                 return r;
2165
2166         sd_bus_close(bus);
2167
2168         bus->current = m;
2169         bus->iteration_counter++;
2170
2171         r = process_filter(bus, m);
2172         if (r != 0)
2173                 goto finish;
2174
2175         r = process_match(bus, m);
2176         if (r != 0)
2177                 goto finish;
2178
2179         if (ret) {
2180                 *ret = m;
2181                 m = NULL;
2182         }
2183
2184         r = 1;
2185
2186 finish:
2187         bus->current = NULL;
2188         return r;
2189 }
2190
2191 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
2192         BUS_DONT_DESTROY(bus);
2193         int r;
2194
2195         /* Returns 0 when we didn't do anything. This should cause the
2196          * caller to invoke sd_bus_wait() before returning the next
2197          * time. Returns > 0 when we did something, which possibly
2198          * means *ret is filled in with an unprocessed message. */
2199
2200         assert_return(bus, -EINVAL);
2201         assert_return(!bus_pid_changed(bus), -ECHILD);
2202
2203         /* We don't allow recursively invoking sd_bus_process(). */
2204         assert_return(!bus->current, -EBUSY);
2205
2206         switch (bus->state) {
2207
2208         case BUS_UNSET:
2209         case BUS_CLOSED:
2210                 return -ENOTCONN;
2211
2212         case BUS_OPENING:
2213                 r = bus_socket_process_opening(bus);
2214                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2215                         bus_enter_closing(bus);
2216                         r = 1;
2217                 } else if (r < 0)
2218                         return r;
2219                 if (ret)
2220                         *ret = NULL;
2221                 return r;
2222
2223         case BUS_AUTHENTICATING:
2224                 r = bus_socket_process_authenticating(bus);
2225                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2226                         bus_enter_closing(bus);
2227                         r = 1;
2228                 } else if (r < 0)
2229                         return r;
2230
2231                 if (ret)
2232                         *ret = NULL;
2233
2234                 return r;
2235
2236         case BUS_RUNNING:
2237         case BUS_HELLO:
2238                 r = process_running(bus, ret);
2239                 if (r == -ECONNRESET || r == -EPIPE || r == -ESHUTDOWN) {
2240                         bus_enter_closing(bus);
2241                         r = 1;
2242
2243                         if (ret)
2244                                 *ret = NULL;
2245                 }
2246
2247                 return r;
2248
2249         case BUS_CLOSING:
2250                 return process_closing(bus, ret);
2251         }
2252
2253         assert_not_reached("Unknown state");
2254 }
2255
2256 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
2257         struct pollfd p[2] = {};
2258         int r, e, n;
2259         struct timespec ts;
2260         usec_t m = (usec_t) -1;
2261
2262         assert(bus);
2263
2264         if (bus->state == BUS_CLOSING)
2265                 return 1;
2266
2267         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2268
2269         e = sd_bus_get_events(bus);
2270         if (e < 0)
2271                 return e;
2272
2273         if (need_more)
2274                 /* The caller really needs some more data, he doesn't
2275                  * care about what's already read, or any timeouts
2276                  * except its own.*/
2277                 e |= POLLIN;
2278         else {
2279                 usec_t until;
2280                 /* The caller wants to process if there's something to
2281                  * process, but doesn't care otherwise */
2282
2283                 r = sd_bus_get_timeout(bus, &until);
2284                 if (r < 0)
2285                         return r;
2286                 if (r > 0) {
2287                         usec_t nw;
2288                         nw = now(CLOCK_MONOTONIC);
2289                         m = until > nw ? until - nw : 0;
2290                 }
2291         }
2292
2293         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
2294                 m = timeout_usec;
2295
2296         p[0].fd = bus->input_fd;
2297         if (bus->output_fd == bus->input_fd) {
2298                 p[0].events = e;
2299                 n = 1;
2300         } else {
2301                 p[0].events = e & POLLIN;
2302                 p[1].fd = bus->output_fd;
2303                 p[1].events = e & POLLOUT;
2304                 n = 2;
2305         }
2306
2307         r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
2308         if (r < 0)
2309                 return -errno;
2310
2311         return r > 0 ? 1 : 0;
2312 }
2313
2314 _public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
2315
2316         assert_return(bus, -EINVAL);
2317         assert_return(!bus_pid_changed(bus), -ECHILD);
2318
2319         if (bus->state == BUS_CLOSING)
2320                 return 0;
2321
2322         assert_return(BUS_IS_OPEN(bus->state) , -ENOTCONN);
2323
2324         if (bus->rqueue_size > 0)
2325                 return 0;
2326
2327         return bus_poll(bus, false, timeout_usec);
2328 }
2329
2330 _public_ int sd_bus_flush(sd_bus *bus) {
2331         int r;
2332
2333         assert_return(bus, -EINVAL);
2334         assert_return(!bus_pid_changed(bus), -ECHILD);
2335
2336         if (bus->state == BUS_CLOSING)
2337                 return 0;
2338
2339         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2340
2341         r = bus_ensure_running(bus);
2342         if (r < 0)
2343                 return r;
2344
2345         if (bus->wqueue_size <= 0)
2346                 return 0;
2347
2348         for (;;) {
2349                 r = dispatch_wqueue(bus);
2350                 if (r < 0) {
2351                         if (r == -EPIPE || r == -ENOTCONN || r == -ESHUTDOWN)
2352                                 bus_enter_closing(bus);
2353
2354                         return r;
2355                 }
2356
2357                 if (bus->wqueue_size <= 0)
2358                         return 0;
2359
2360                 r = bus_poll(bus, false, (uint64_t) -1);
2361                 if (r < 0)
2362                         return r;
2363         }
2364 }
2365
2366 _public_ int sd_bus_add_filter(sd_bus *bus,
2367                                sd_bus_message_handler_t callback,
2368                                void *userdata) {
2369
2370         struct filter_callback *f;
2371
2372         assert_return(bus, -EINVAL);
2373         assert_return(callback, -EINVAL);
2374         assert_return(!bus_pid_changed(bus), -ECHILD);
2375
2376         f = new0(struct filter_callback, 1);
2377         if (!f)
2378                 return -ENOMEM;
2379         f->callback = callback;
2380         f->userdata = userdata;
2381
2382         bus->filter_callbacks_modified = true;
2383         LIST_PREPEND(callbacks, bus->filter_callbacks, f);
2384         return 0;
2385 }
2386
2387 _public_ int sd_bus_remove_filter(sd_bus *bus,
2388                                   sd_bus_message_handler_t callback,
2389                                   void *userdata) {
2390
2391         struct filter_callback *f;
2392
2393         assert_return(bus, -EINVAL);
2394         assert_return(callback, -EINVAL);
2395         assert_return(!bus_pid_changed(bus), -ECHILD);
2396
2397         LIST_FOREACH(callbacks, f, bus->filter_callbacks) {
2398                 if (f->callback == callback && f->userdata == userdata) {
2399                         bus->filter_callbacks_modified = true;
2400                         LIST_REMOVE(callbacks, bus->filter_callbacks, f);
2401                         free(f);
2402                         return 1;
2403                 }
2404         }
2405
2406         return 0;
2407 }
2408
2409 _public_ int sd_bus_add_match(sd_bus *bus,
2410                               const char *match,
2411                               sd_bus_message_handler_t callback,
2412                               void *userdata) {
2413
2414         struct bus_match_component *components = NULL;
2415         unsigned n_components = 0;
2416         uint64_t cookie = 0;
2417         int r = 0;
2418
2419         assert_return(bus, -EINVAL);
2420         assert_return(match, -EINVAL);
2421         assert_return(!bus_pid_changed(bus), -ECHILD);
2422
2423         r = bus_match_parse(match, &components, &n_components);
2424         if (r < 0)
2425                 goto finish;
2426
2427         if (bus->bus_client) {
2428                 cookie = ++bus->match_cookie;
2429
2430                 r = bus_add_match_internal(bus, match, components, n_components, cookie);
2431                 if (r < 0)
2432                         goto finish;
2433         }
2434
2435         bus->match_callbacks_modified = true;
2436         r = bus_match_add(&bus->match_callbacks, components, n_components, callback, userdata, cookie, NULL);
2437         if (r < 0) {
2438                 if (bus->bus_client)
2439                         bus_remove_match_internal(bus, match, cookie);
2440         }
2441
2442 finish:
2443         bus_match_parse_free(components, n_components);
2444         return r;
2445 }
2446
2447 _public_ int sd_bus_remove_match(sd_bus *bus,
2448                                  const char *match,
2449                                  sd_bus_message_handler_t callback,
2450                                  void *userdata) {
2451
2452         struct bus_match_component *components = NULL;
2453         unsigned n_components = 0;
2454         int r = 0, q = 0;
2455         uint64_t cookie = 0;
2456
2457         assert_return(bus, -EINVAL);
2458         assert_return(match, -EINVAL);
2459         assert_return(!bus_pid_changed(bus), -ECHILD);
2460
2461         r = bus_match_parse(match, &components, &n_components);
2462         if (r < 0)
2463                 return r;
2464
2465         bus->match_callbacks_modified = true;
2466         r = bus_match_remove(&bus->match_callbacks, components, n_components, callback, userdata, &cookie);
2467
2468         if (bus->bus_client)
2469                 q = bus_remove_match_internal(bus, match, cookie);
2470
2471         bus_match_parse_free(components, n_components);
2472
2473         return r < 0 ? r : q;
2474 }
2475
2476 bool bus_pid_changed(sd_bus *bus) {
2477         assert(bus);
2478
2479         /* We don't support people creating a bus connection and
2480          * keeping it around over a fork(). Let's complain. */
2481
2482         return bus->original_pid != getpid();
2483 }
2484
2485 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
2486         sd_bus *bus = userdata;
2487         int r;
2488
2489         assert(bus);
2490
2491         r = sd_bus_process(bus, NULL);
2492         if (r < 0)
2493                 return r;
2494
2495         return 1;
2496 }
2497
2498 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
2499         sd_bus *bus = userdata;
2500         int r;
2501
2502         assert(bus);
2503
2504         r = sd_bus_process(bus, NULL);
2505         if (r < 0)
2506                 return r;
2507
2508         return 1;
2509 }
2510
2511 static int prepare_callback(sd_event_source *s, void *userdata) {
2512         sd_bus *bus = userdata;
2513         int r, e;
2514         usec_t until;
2515
2516         assert(s);
2517         assert(bus);
2518
2519         e = sd_bus_get_events(bus);
2520         if (e < 0)
2521                 return e;
2522
2523         if (bus->output_fd != bus->input_fd) {
2524
2525                 r = sd_event_source_set_io_events(bus->input_io_event_source, e & POLLIN);
2526                 if (r < 0)
2527                         return r;
2528
2529                 r = sd_event_source_set_io_events(bus->output_io_event_source, e & POLLOUT);
2530                 if (r < 0)
2531                         return r;
2532         } else {
2533                 r = sd_event_source_set_io_events(bus->input_io_event_source, e);
2534                 if (r < 0)
2535                         return r;
2536         }
2537
2538         r = sd_bus_get_timeout(bus, &until);
2539         if (r < 0)
2540                 return r;
2541         if (r > 0) {
2542                 int j;
2543
2544                 j = sd_event_source_set_time(bus->time_event_source, until);
2545                 if (j < 0)
2546                         return j;
2547         }
2548
2549         r = sd_event_source_set_enabled(bus->time_event_source, r > 0);
2550         if (r < 0)
2551                 return r;
2552
2553         return 1;
2554 }
2555
2556 static int quit_callback(sd_event_source *event, void *userdata) {
2557         sd_bus *bus = userdata;
2558
2559         assert(event);
2560
2561         sd_bus_flush(bus);
2562
2563         return 1;
2564 }
2565
2566 _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
2567         int r;
2568
2569         assert_return(bus, -EINVAL);
2570         assert_return(!bus->event, -EBUSY);
2571
2572         assert(!bus->input_io_event_source);
2573         assert(!bus->output_io_event_source);
2574         assert(!bus->time_event_source);
2575
2576         if (event)
2577                 bus->event = sd_event_ref(event);
2578         else  {
2579                 r = sd_event_default(&bus->event);
2580                 if (r < 0)
2581                         return r;
2582         }
2583
2584         r = sd_event_add_io(bus->event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
2585         if (r < 0)
2586                 goto fail;
2587
2588         r = sd_event_source_set_priority(bus->input_io_event_source, priority);
2589         if (r < 0)
2590                 goto fail;
2591
2592         if (bus->output_fd != bus->input_fd) {
2593                 r = sd_event_add_io(bus->event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
2594                 if (r < 0)
2595                         goto fail;
2596
2597                 r = sd_event_source_set_priority(bus->output_io_event_source, priority);
2598                 if (r < 0)
2599                         goto fail;
2600         }
2601
2602         r = sd_event_source_set_prepare(bus->input_io_event_source, prepare_callback);
2603         if (r < 0)
2604                 goto fail;
2605
2606         r = sd_event_add_monotonic(bus->event, 0, 0, time_callback, bus, &bus->time_event_source);
2607         if (r < 0)
2608                 goto fail;
2609
2610         r = sd_event_source_set_priority(bus->time_event_source, priority);
2611         if (r < 0)
2612                 goto fail;
2613
2614         r = sd_event_add_quit(bus->event, quit_callback, bus, &bus->quit_event_source);
2615         if (r < 0)
2616                 goto fail;
2617
2618         return 0;
2619
2620 fail:
2621         sd_bus_detach_event(bus);
2622         return r;
2623 }
2624
2625 _public_ int sd_bus_detach_event(sd_bus *bus) {
2626         assert_return(bus, -EINVAL);
2627         assert_return(bus->event, -ENXIO);
2628
2629         if (bus->input_io_event_source) {
2630                 sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
2631                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
2632         }
2633
2634         if (bus->output_io_event_source) {
2635                 sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
2636                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
2637         }
2638
2639         if (bus->time_event_source) {
2640                 sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
2641                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
2642         }
2643
2644         if (bus->quit_event_source) {
2645                 sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
2646                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
2647         }
2648
2649         if (bus->event)
2650                 bus->event = sd_event_unref(bus->event);
2651
2652         return 0;
2653 }
2654
2655 _public_ sd_event* sd_bus_get_event(sd_bus *bus) {
2656         assert_return(bus, NULL);
2657
2658         return bus->event;
2659 }
2660
2661 _public_ sd_bus_message* sd_bus_get_current(sd_bus *bus) {
2662         assert_return(bus, NULL);
2663
2664         return bus->current;
2665 }
2666
2667 static int bus_default(int (*bus_open)(sd_bus **), sd_bus **default_bus, sd_bus **ret) {
2668         sd_bus *b = NULL;
2669         int r;
2670
2671         assert(bus_open);
2672         assert(default_bus);
2673
2674         if (!ret)
2675                 return !!*default_bus;
2676
2677         if (*default_bus) {
2678                 *ret = sd_bus_ref(*default_bus);
2679                 return 0;
2680         }
2681
2682         r = bus_open(&b);
2683         if (r < 0)
2684                 return r;
2685
2686         b->default_bus_ptr = default_bus;
2687         b->tid = gettid();
2688         *default_bus = b;
2689
2690         *ret = b;
2691         return 1;
2692 }
2693
2694 _public_ int sd_bus_default_system(sd_bus **ret) {
2695         static __thread sd_bus *default_system_bus = NULL;
2696
2697         return bus_default(sd_bus_open_system, &default_system_bus, ret);
2698 }
2699
2700 _public_ int sd_bus_default_user(sd_bus **ret) {
2701         static __thread sd_bus *default_user_bus = NULL;
2702
2703         return bus_default(sd_bus_open_user, &default_user_bus, ret);
2704 }
2705
2706 _public_ int sd_bus_get_tid(sd_bus *b, pid_t *tid) {
2707         assert_return(b, -EINVAL);
2708         assert_return(tid, -EINVAL);
2709         assert_return(!bus_pid_changed(b), -ECHILD);
2710
2711         if (b->tid != 0) {
2712                 *tid = b->tid;
2713                 return 0;
2714         }
2715
2716         if (b->event)
2717                 return sd_event_get_tid(b->event, tid);
2718
2719         return -ENXIO;
2720 }
2721
2722 _public_ char *sd_bus_label_escape(const char *s) {
2723         char *r, *t;
2724         const char *f;
2725
2726         assert_return(s, NULL);
2727
2728         /* Escapes all chars that D-Bus' object path cannot deal
2729          * with. Can be reversed with bus_path_unescape(). We special
2730          * case the empty string. */
2731
2732         if (*s == 0)
2733                 return strdup("_");
2734
2735         r = new(char, strlen(s)*3 + 1);
2736         if (!r)
2737                 return NULL;
2738
2739         for (f = s, t = r; *f; f++) {
2740
2741                 /* Escape everything that is not a-zA-Z0-9. We also
2742                  * escape 0-9 if it's the first character */
2743
2744                 if (!(*f >= 'A' && *f <= 'Z') &&
2745                     !(*f >= 'a' && *f <= 'z') &&
2746                     !(f > s && *f >= '0' && *f <= '9')) {
2747                         *(t++) = '_';
2748                         *(t++) = hexchar(*f >> 4);
2749                         *(t++) = hexchar(*f);
2750                 } else
2751                         *(t++) = *f;
2752         }
2753
2754         *t = 0;
2755
2756         return r;
2757 }
2758
2759 _public_ char *sd_bus_label_unescape(const char *f) {
2760         char *r, *t;
2761
2762         assert_return(f, NULL);
2763
2764         /* Special case for the empty string */
2765         if (streq(f, "_"))
2766                 return strdup("");
2767
2768         r = new(char, strlen(f) + 1);
2769         if (!r)
2770                 return NULL;
2771
2772         for (t = r; *f; f++) {
2773
2774                 if (*f == '_') {
2775                         int a, b;
2776
2777                         if ((a = unhexchar(f[1])) < 0 ||
2778                             (b = unhexchar(f[2])) < 0) {
2779                                 /* Invalid escape code, let's take it literal then */
2780                                 *(t++) = '_';
2781                         } else {
2782                                 *(t++) = (char) ((a << 4) | b);
2783                                 f += 2;
2784                         }
2785                 } else
2786                         *(t++) = *f;
2787         }
2788
2789         *t = 0;
2790
2791         return r;
2792 }
2793
2794 _public_ int sd_bus_get_peer_creds(sd_bus *bus, uint64_t mask, sd_bus_creds **ret) {
2795         sd_bus_creds *c;
2796         pid_t pid = 0;
2797         int r;
2798
2799         assert_return(bus, -EINVAL);
2800         assert_return(mask <= _SD_BUS_CREDS_ALL, -ENOTSUP);
2801         assert_return(ret, -EINVAL);
2802         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
2803         assert_return(!bus_pid_changed(bus), -ECHILD);
2804         assert_return(!bus->is_kernel, -ENOTSUP);
2805
2806         if (!bus->ucred_valid && !isempty(bus->label))
2807                 return -ENODATA;
2808
2809         c = bus_creds_new();
2810         if (!c)
2811                 return -ENOMEM;
2812
2813         if (bus->ucred_valid) {
2814                 pid = c->pid = bus->ucred.pid;
2815                 c->uid = bus->ucred.uid;
2816                 c->gid = bus->ucred.gid;
2817
2818                 c->mask |= (SD_BUS_CREDS_UID | SD_BUS_CREDS_PID | SD_BUS_CREDS_GID) & mask;
2819         }
2820
2821         if (!isempty(bus->label) && (mask & SD_BUS_CREDS_SELINUX_CONTEXT)) {
2822                 c->label = strdup(bus->label);
2823                 if (!c->label) {
2824                         sd_bus_creds_unref(c);
2825                         return -ENOMEM;
2826                 }
2827
2828                 c->mask |= SD_BUS_CREDS_SELINUX_CONTEXT;
2829         }
2830
2831         r = bus_creds_add_more(c, mask, pid, 0);
2832         if (r < 0)
2833                 return r;
2834
2835         *ret = c;
2836         return 0;
2837 }