src/libsystemd-bus/sd-bus.c

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