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