src/libsystemd-bus/sd-bus.c

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