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
  49 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec);
  50
  51 static void bus_close_fds(sd_bus *b) {
  52         assert(b);
  53
  54         if (b->input_fd >= 0)
  55                 close_nointr_nofail(b->input_fd);
  56
  57         if (b->output_fd >= 0 && b->output_fd != b->input_fd)
  58                 close_nointr_nofail(b->output_fd);
  59
  60         b->input_fd = b->output_fd = -1;
  61 }
  62
  63 static void bus_node_destroy(sd_bus *b, struct node *n) {
  64         struct node_callback *c;
  65         struct node_vtable *v;
  66         struct node_enumerator *e;
  67
  68         assert(b);
  69
  70         if (!n)
  71                 return;
  72
  73         while (n->child)
  74                 bus_node_destroy(b, n->child);
  75
  76         while ((c = n->callbacks)) {
  77                 LIST_REMOVE(callbacks, n->callbacks, c);
  78                 free(c);
  79         }
  80
  81         while ((v = n->vtables)) {
  82                 LIST_REMOVE(vtables, n->vtables, v);
  83                 free(v->interface);
  84                 free(v);
  85         }
  86
  87         while ((e = n->enumerators)) {
  88                 LIST_REMOVE(enumerators, n->enumerators, e);
  89                 free(e);
  90         }
  91
  92         if (n->parent)
  93                 LIST_REMOVE(siblings, n->parent->child, n);
  94
  95         assert_se(hashmap_remove(b->nodes, n->path) == n);
  96         free(n->path);
  97         free(n);
  98 }
  99
 100 static void bus_free(sd_bus *b) {
 101         struct filter_callback *f;
 102         struct node *n;
 103         unsigned i;
 104
 105         assert(b);
 106
 107         bus_close_fds(b);
 108
 109         if (b->kdbus_buffer)
 110                 munmap(b->kdbus_buffer, KDBUS_POOL_SIZE);
 111
 112         free(b->rbuffer);
 113         free(b->unique_name);
 114         free(b->auth_buffer);
 115         free(b->address);
 116         free(b->kernel);
 117
 118         free(b->exec_path);
 119         strv_free(b->exec_argv);
 120
 121         close_many(b->fds, b->n_fds);
 122         free(b->fds);
 123
 124         for (i = 0; i < b->rqueue_size; i++)
 125                 sd_bus_message_unref(b->rqueue[i]);
 126         free(b->rqueue);
 127
 128         for (i = 0; i < b->wqueue_size; i++)
 129                 sd_bus_message_unref(b->wqueue[i]);
 130         free(b->wqueue);
 131
 132         hashmap_free_free(b->reply_callbacks);
 133         prioq_free(b->reply_callbacks_prioq);
 134
 135         while ((f = b->filter_callbacks)) {
 136                 LIST_REMOVE(callbacks, b->filter_callbacks, f);
 137                 free(f);
 138         }
 139
 140         bus_match_free(&b->match_callbacks);
 141
 142         hashmap_free_free(b->vtable_methods);
 143         hashmap_free_free(b->vtable_properties);
 144
 145         while ((n = hashmap_first(b->nodes)))
 146                 bus_node_destroy(b, n);
 147
 148         hashmap_free(b->nodes);
 149
 150         bus_kernel_flush_memfd(b);
 151
 152         assert_se(pthread_mutex_destroy(&b->memfd_cache_mutex) == 0);
 153
 154         free(b);
 155 }
 156
 157 int sd_bus_new(sd_bus **ret) {
 158         sd_bus *r;
 159
 160         assert_return(ret, -EINVAL);
 161
 162         r = new0(sd_bus, 1);
 163         if (!r)
 164                 return -ENOMEM;
 165
 166         r->n_ref = REFCNT_INIT;
 167         r->input_fd = r->output_fd = -1;
 168         r->message_version = 1;
 169         r->hello_flags |= KDBUS_HELLO_ACCEPT_FD;
 170         r->original_pid = getpid();
 171
 172         assert_se(pthread_mutex_init(&r->memfd_cache_mutex, NULL) == 0);
 173
 174         /* We guarantee that wqueue always has space for at least one
 175          * entry */
 176         r->wqueue = new(sd_bus_message*, 1);
 177         if (!r->wqueue) {
 178                 free(r);
 179                 return -ENOMEM;
 180         }
 181
 182         *ret = r;
 183         return 0;
 184 }
 185
 186 int sd_bus_set_address(sd_bus *bus, const char *address) {
 187         char *a;
 188
 189         assert_return(bus, -EINVAL);
 190         assert_return(bus->state == BUS_UNSET, -EPERM);
 191         assert_return(address, -EINVAL);
 192         assert_return(!bus_pid_changed(bus), -ECHILD);
 193
 194         a = strdup(address);
 195         if (!a)
 196                 return -ENOMEM;
 197
 198         free(bus->address);
 199         bus->address = a;
 200
 201         return 0;
 202 }
 203
 204 int sd_bus_set_fd(sd_bus *bus, int input_fd, int output_fd) {
 205         assert_return(bus, -EINVAL);
 206         assert_return(bus->state == BUS_UNSET, -EPERM);
 207         assert_return(input_fd >= 0, -EINVAL);
 208         assert_return(output_fd >= 0, -EINVAL);
 209         assert_return(!bus_pid_changed(bus), -ECHILD);
 210
 211         bus->input_fd = input_fd;
 212         bus->output_fd = output_fd;
 213         return 0;
 214 }
 215
 216 int sd_bus_set_exec(sd_bus *bus, const char *path, char *const argv[]) {
 217         char *p, **a;
 218
 219         assert_return(bus, -EINVAL);
 220         assert_return(bus->state == BUS_UNSET, -EPERM);
 221         assert_return(path, -EINVAL);
 222         assert_return(!strv_isempty(argv), -EINVAL);
 223         assert_return(!bus_pid_changed(bus), -ECHILD);
 224
 225         p = strdup(path);
 226         if (!p)
 227                 return -ENOMEM;
 228
 229         a = strv_copy(argv);
 230         if (!a) {
 231                 free(p);
 232                 return -ENOMEM;
 233         }
 234
 235         free(bus->exec_path);
 236         strv_free(bus->exec_argv);
 237
 238         bus->exec_path = p;
 239         bus->exec_argv = a;
 240
 241         return 0;
 242 }
 243
 244 int sd_bus_set_bus_client(sd_bus *bus, int b) {
 245         assert_return(bus, -EINVAL);
 246         assert_return(bus->state == BUS_UNSET, -EPERM);
 247         assert_return(!bus_pid_changed(bus), -ECHILD);
 248
 249         bus->bus_client = !!b;
 250         return 0;
 251 }
 252
 253 int sd_bus_negotiate_fds(sd_bus *bus, int b) {
 254         assert_return(bus, -EINVAL);
 255         assert_return(bus->state == BUS_UNSET, -EPERM);
 256         assert_return(!bus_pid_changed(bus), -ECHILD);
 257
 258         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ACCEPT_FD, b);
 259         return 0;
 260 }
 261
 262 int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
 263         assert_return(bus, -EINVAL);
 264         assert_return(bus->state == BUS_UNSET, -EPERM);
 265         assert_return(!bus_pid_changed(bus), -ECHILD);
 266
 267         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_COMM, b);
 268         return 0;
 269 }
 270
 271 int sd_bus_negotiate_attach_exe(sd_bus *bus, int b) {
 272         assert_return(bus, -EINVAL);
 273         assert_return(bus->state == BUS_UNSET, -EPERM);
 274         assert_return(!bus_pid_changed(bus), -ECHILD);
 275
 276         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_EXE, b);
 277         return 0;
 278 }
 279
 280 int sd_bus_negotiate_attach_cmdline(sd_bus *bus, int b) {
 281         assert_return(bus, -EINVAL);
 282         assert_return(bus->state == BUS_UNSET, -EPERM);
 283         assert_return(!bus_pid_changed(bus), -ECHILD);
 284
 285         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CMDLINE, b);
 286         return 0;
 287 }
 288
 289 int sd_bus_negotiate_attach_cgroup(sd_bus *bus, int b) {
 290         assert_return(bus, -EINVAL);
 291         assert_return(bus->state == BUS_UNSET, -EPERM);
 292         assert_return(!bus_pid_changed(bus), -ECHILD);
 293
 294         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CGROUP, b);
 295         return 0;
 296 }
 297
 298 int sd_bus_negotiate_attach_caps(sd_bus *bus, int b) {
 299         assert_return(bus, -EINVAL);
 300         assert_return(bus->state == BUS_UNSET, -EPERM);
 301         assert_return(!bus_pid_changed(bus), -ECHILD);
 302
 303         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CAPS, b);
 304         return 0;
 305 }
 306
 307 int sd_bus_negotiate_attach_selinux_context(sd_bus *bus, int b) {
 308         assert_return(bus, -EINVAL);
 309         assert_return(bus->state == BUS_UNSET, -EPERM);
 310         assert_return(!bus_pid_changed(bus), -ECHILD);
 311
 312         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_SECLABEL, b);
 313         return 0;
 314 }
 315
 316 int sd_bus_negotiate_attach_audit(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         SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_AUDIT, b);
 322         return 0;
 323 }
 324
 325 int sd_bus_set_server(sd_bus *bus, int b, sd_id128_t server_id) {
 326         assert_return(bus, -EINVAL);
 327         assert_return(b || sd_id128_equal(server_id, SD_ID128_NULL), -EINVAL);
 328         assert_return(bus->state == BUS_UNSET, -EPERM);
 329         assert_return(!bus_pid_changed(bus), -ECHILD);
 330
 331         bus->is_server = !!b;
 332         bus->server_id = server_id;
 333         return 0;
 334 }
 335
 336 int sd_bus_set_anonymous(sd_bus *bus, int b) {
 337         assert_return(bus, -EINVAL);
 338         assert_return(bus->state == BUS_UNSET, -EPERM);
 339         assert_return(!bus_pid_changed(bus), -ECHILD);
 340
 341         bus->anonymous_auth = !!b;
 342         return 0;
 343 }
 344
 345 static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
 346         const char *s;
 347         int r;
 348
 349         assert(bus);
 350         assert(bus->state == BUS_HELLO);
 351         assert(reply);
 352
 353         r = bus_message_to_errno(reply);
 354         if (r < 0)
 355                 return r;
 356
 357         r = sd_bus_message_read(reply, "s", &s);
 358         if (r < 0)
 359                 return r;
 360
 361         if (!service_name_is_valid(s) || s[0] != ':')
 362                 return -EBADMSG;
 363
 364         bus->unique_name = strdup(s);
 365         if (!bus->unique_name)
 366                 return -ENOMEM;
 367
 368         bus->state = BUS_RUNNING;
 369
 370         return 1;
 371 }
 372
 373 static int bus_send_hello(sd_bus *bus) {
 374         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
 375         int r;
 376
 377         assert(bus);
 378
 379         if (!bus->bus_client || bus->is_kernel)
 380                 return 0;
 381
 382         r = sd_bus_message_new_method_call(
 383                         bus,
 384                         "org.freedesktop.DBus",
 385                         "/",
 386                         "org.freedesktop.DBus",
 387                         "Hello",
 388                         &m);
 389         if (r < 0)
 390                 return r;
 391
 392         return sd_bus_send_with_reply(bus, m, hello_callback, NULL, 0, &bus->hello_serial);
 393 }
 394
 395 int bus_start_running(sd_bus *bus) {
 396         assert(bus);
 397
 398         if (bus->bus_client && !bus->is_kernel) {
 399                 bus->state = BUS_HELLO;
 400                 return 1;
 401         }
 402
 403         bus->state = BUS_RUNNING;
 404         return 1;
 405 }
 406
 407 static int parse_address_key(const char **p, const char *key, char **value) {
 408         size_t l, n = 0;
 409         const char *a;
 410         char *r = NULL;
 411
 412         assert(p);
 413         assert(*p);
 414         assert(value);
 415
 416         if (key) {
 417                 l = strlen(key);
 418                 if (strncmp(*p, key, l) != 0)
 419                         return 0;
 420
 421                 if ((*p)[l] != '=')
 422                         return 0;
 423
 424                 if (*value)
 425                         return -EINVAL;
 426
 427                 a = *p + l + 1;
 428         } else
 429                 a = *p;
 430
 431         while (*a != ';' && *a != ',' && *a != 0) {
 432                 char c, *t;
 433
 434                 if (*a == '%') {
 435                         int x, y;
 436
 437                         x = unhexchar(a[1]);
 438                         if (x < 0) {
 439                                 free(r);
 440                                 return x;
 441                         }
 442
 443                         y = unhexchar(a[2]);
 444                         if (y < 0) {
 445                                 free(r);
 446                                 return y;
 447                         }
 448
 449                         c = (char) ((x << 4) | y);
 450                         a += 3;
 451                 } else {
 452                         c = *a;
 453                         a++;
 454                 }
 455
 456                 t = realloc(r, n + 2);
 457                 if (!t) {
 458                         free(r);
 459                         return -ENOMEM;
 460                 }
 461
 462                 r = t;
 463                 r[n++] = c;
 464         }
 465
 466         if (!r) {
 467                 r = strdup("");
 468                 if (!r)
 469                         return -ENOMEM;
 470         } else
 471                 r[n] = 0;
 472
 473         if (*a == ',')
 474                 a++;
 475
 476         *p = a;
 477
 478         free(*value);
 479         *value = r;
 480
 481         return 1;
 482 }
 483
 484 static void skip_address_key(const char **p) {
 485         assert(p);
 486         assert(*p);
 487
 488         *p += strcspn(*p, ",");
 489
 490         if (**p == ',')
 491                 (*p) ++;
 492 }
 493
 494 static int parse_unix_address(sd_bus *b, const char **p, char **guid) {
 495         _cleanup_free_ char *path = NULL, *abstract = NULL;
 496         size_t l;
 497         int r;
 498
 499         assert(b);
 500         assert(p);
 501         assert(*p);
 502         assert(guid);
 503
 504         while (**p != 0 && **p != ';') {
 505                 r = parse_address_key(p, "guid", guid);
 506                 if (r < 0)
 507                         return r;
 508                 else if (r > 0)
 509                         continue;
 510
 511                 r = parse_address_key(p, "path", &path);
 512                 if (r < 0)
 513                         return r;
 514                 else if (r > 0)
 515                         continue;
 516
 517                 r = parse_address_key(p, "abstract", &abstract);
 518                 if (r < 0)
 519                         return r;
 520                 else if (r > 0)
 521                         continue;
 522
 523                 skip_address_key(p);
 524         }
 525
 526         if (!path && !abstract)
 527                 return -EINVAL;
 528
 529         if (path && abstract)
 530                 return -EINVAL;
 531
 532         if (path) {
 533                 l = strlen(path);
 534                 if (l > sizeof(b->sockaddr.un.sun_path))
 535                         return -E2BIG;
 536
 537                 b->sockaddr.un.sun_family = AF_UNIX;
 538                 strncpy(b->sockaddr.un.sun_path, path, sizeof(b->sockaddr.un.sun_path));
 539                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l;
 540         } else if (abstract) {
 541                 l = strlen(abstract);
 542                 if (l > sizeof(b->sockaddr.un.sun_path) - 1)
 543                         return -E2BIG;
 544
 545                 b->sockaddr.un.sun_family = AF_UNIX;
 546                 b->sockaddr.un.sun_path[0] = 0;
 547                 strncpy(b->sockaddr.un.sun_path+1, abstract, sizeof(b->sockaddr.un.sun_path)-1);
 548                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + 1 + l;
 549         }
 550
 551         return 0;
 552 }
 553
 554 static int parse_tcp_address(sd_bus *b, const char **p, char **guid) {
 555         _cleanup_free_ char *host = NULL, *port = NULL, *family = NULL;
 556         int r;
 557         struct addrinfo *result, hints = {
 558                 .ai_socktype = SOCK_STREAM,
 559                 .ai_flags = AI_ADDRCONFIG,
 560         };
 561
 562         assert(b);
 563         assert(p);
 564         assert(*p);
 565         assert(guid);
 566
 567         while (**p != 0 && **p != ';') {
 568                 r = parse_address_key(p, "guid", guid);
 569                 if (r < 0)
 570                         return r;
 571                 else if (r > 0)
 572                         continue;
 573
 574                 r = parse_address_key(p, "host", &host);
 575                 if (r < 0)
 576                         return r;
 577                 else if (r > 0)
 578                         continue;
 579
 580                 r = parse_address_key(p, "port", &port);
 581                 if (r < 0)
 582                         return r;
 583                 else if (r > 0)
 584                         continue;
 585
 586                 r = parse_address_key(p, "family", &family);
 587                 if (r < 0)
 588                         return r;
 589                 else if (r > 0)
 590                         continue;
 591
 592                 skip_address_key(p);
 593         }
 594
 595         if (!host || !port)
 596                 return -EINVAL;
 597
 598         if (family) {
 599                 if (streq(family, "ipv4"))
 600                         hints.ai_family = AF_INET;
 601                 else if (streq(family, "ipv6"))
 602                         hints.ai_family = AF_INET6;
 603                 else
 604                         return -EINVAL;
 605         }
 606
 607         r = getaddrinfo(host, port, &hints, &result);
 608         if (r == EAI_SYSTEM)
 609                 return -errno;
 610         else if (r != 0)
 611                 return -EADDRNOTAVAIL;
 612
 613         memcpy(&b->sockaddr, result->ai_addr, result->ai_addrlen);
 614         b->sockaddr_size = result->ai_addrlen;
 615
 616         freeaddrinfo(result);
 617
 618         return 0;
 619 }
 620
 621 static int parse_exec_address(sd_bus *b, const char **p, char **guid) {
 622         char *path = NULL;
 623         unsigned n_argv = 0, j;
 624         char **argv = NULL;
 625         int r;
 626
 627         assert(b);
 628         assert(p);
 629         assert(*p);
 630         assert(guid);
 631
 632         while (**p != 0 && **p != ';') {
 633                 r = parse_address_key(p, "guid", guid);
 634                 if (r < 0)
 635                         goto fail;
 636                 else if (r > 0)
 637                         continue;
 638
 639                 r = parse_address_key(p, "path", &path);
 640                 if (r < 0)
 641                         goto fail;
 642                 else if (r > 0)
 643                         continue;
 644
 645                 if (startswith(*p, "argv")) {
 646                         unsigned ul;
 647
 648                         errno = 0;
 649                         ul = strtoul(*p + 4, (char**) p, 10);
 650                         if (errno > 0 || **p != '=' || ul > 256) {
 651                                 r = -EINVAL;
 652                                 goto fail;
 653                         }
 654
 655                         (*p) ++;
 656
 657                         if (ul >= n_argv) {
 658                                 char **x;
 659
 660                                 x = realloc(argv, sizeof(char*) * (ul + 2));
 661                                 if (!x) {
 662                                         r = -ENOMEM;
 663                                         goto fail;
 664                                 }
 665
 666                                 memset(x + n_argv, 0, sizeof(char*) * (ul - n_argv + 2));
 667
 668                                 argv = x;
 669                                 n_argv = ul + 1;
 670                         }
 671
 672                         r = parse_address_key(p, NULL, argv + ul);
 673                         if (r < 0)
 674                                 goto fail;
 675
 676                         continue;
 677                 }
 678
 679                 skip_address_key(p);
 680         }
 681
 682         if (!path) {
 683                 r = -EINVAL;
 684                 goto fail;
 685         }
 686
 687         /* Make sure there are no holes in the array, with the
 688          * exception of argv[0] */
 689         for (j = 1; j < n_argv; j++)
 690                 if (!argv[j]) {
 691                         r = -EINVAL;
 692                         goto fail;
 693                 }
 694
 695         if (argv && argv[0] == NULL) {
 696                 argv[0] = strdup(path);
 697                 if (!argv[0]) {
 698                         r = -ENOMEM;
 699                         goto fail;
 700                 }
 701         }
 702
 703         b->exec_path = path;
 704         b->exec_argv = argv;
 705         return 0;
 706
 707 fail:
 708         for (j = 0; j < n_argv; j++)
 709                 free(argv[j]);
 710
 711         free(argv);
 712         free(path);
 713         return r;
 714 }
 715
 716 static int parse_kernel_address(sd_bus *b, const char **p, char **guid) {
 717         _cleanup_free_ char *path = NULL;
 718         int r;
 719
 720         assert(b);
 721         assert(p);
 722         assert(*p);
 723         assert(guid);
 724
 725         while (**p != 0 && **p != ';') {
 726                 r = parse_address_key(p, "guid", guid);
 727                 if (r < 0)
 728                         return r;
 729                 else if (r > 0)
 730                         continue;
 731
 732                 r = parse_address_key(p, "path", &path);
 733                 if (r < 0)
 734                         return r;
 735                 else if (r > 0)
 736                         continue;
 737
 738                 skip_address_key(p);
 739         }
 740
 741         if (!path)
 742                 return -EINVAL;
 743
 744         free(b->kernel);
 745         b->kernel = path;
 746         path = NULL;
 747
 748         return 0;
 749 }
 750
 751 static void bus_reset_parsed_address(sd_bus *b) {
 752         assert(b);
 753
 754         zero(b->sockaddr);
 755         b->sockaddr_size = 0;
 756         strv_free(b->exec_argv);
 757         free(b->exec_path);
 758         b->exec_path = NULL;
 759         b->exec_argv = NULL;
 760         b->server_id = SD_ID128_NULL;
 761         free(b->kernel);
 762         b->kernel = NULL;
 763 }
 764
 765 static int bus_parse_next_address(sd_bus *b) {
 766         _cleanup_free_ char *guid = NULL;
 767         const char *a;
 768         int r;
 769
 770         assert(b);
 771
 772         if (!b->address)
 773                 return 0;
 774         if (b->address[b->address_index] == 0)
 775                 return 0;
 776
 777         bus_reset_parsed_address(b);
 778
 779         a = b->address + b->address_index;
 780
 781         while (*a != 0) {
 782
 783                 if (*a == ';') {
 784                         a++;
 785                         continue;
 786                 }
 787
 788                 if (startswith(a, "unix:")) {
 789                         a += 5;
 790
 791                         r = parse_unix_address(b, &a, &guid);
 792                         if (r < 0)
 793                                 return r;
 794                         break;
 795
 796                 } else if (startswith(a, "tcp:")) {
 797
 798                         a += 4;
 799                         r = parse_tcp_address(b, &a, &guid);
 800                         if (r < 0)
 801                                 return r;
 802
 803                         break;
 804
 805                 } else if (startswith(a, "unixexec:")) {
 806
 807                         a += 9;
 808                         r = parse_exec_address(b, &a, &guid);
 809                         if (r < 0)
 810                                 return r;
 811
 812                         break;
 813
 814                 } else if (startswith(a, "kernel:")) {
 815
 816                         a += 7;
 817                         r = parse_kernel_address(b, &a, &guid);
 818                         if (r < 0)
 819                                 return r;
 820
 821                         break;
 822                 }
 823
 824                 a = strchr(a, ';');
 825                 if (!a)
 826                         return 0;
 827         }
 828
 829         if (guid) {
 830                 r = sd_id128_from_string(guid, &b->server_id);
 831                 if (r < 0)
 832                         return r;
 833         }
 834
 835         b->address_index = a - b->address;
 836         return 1;
 837 }
 838
 839 static int bus_start_address(sd_bus *b) {
 840         int r;
 841
 842         assert(b);
 843
 844         for (;;) {
 845                 sd_bus_close(b);
 846
 847                 if (b->sockaddr.sa.sa_family != AF_UNSPEC) {
 848
 849                         r = bus_socket_connect(b);
 850                         if (r >= 0)
 851                                 return r;
 852
 853                         b->last_connect_error = -r;
 854
 855                 } else if (b->exec_path) {
 856
 857                         r = bus_socket_exec(b);
 858                         if (r >= 0)
 859                                 return r;
 860
 861                         b->last_connect_error = -r;
 862                 } else if (b->kernel) {
 863
 864                         r = bus_kernel_connect(b);
 865                         if (r >= 0)
 866                                 return r;
 867
 868                         b->last_connect_error = -r;
 869                 }
 870
 871                 r = bus_parse_next_address(b);
 872                 if (r < 0)
 873                         return r;
 874                 if (r == 0)
 875                         return b->last_connect_error ? -b->last_connect_error : -ECONNREFUSED;
 876         }
 877 }
 878
 879 int bus_next_address(sd_bus *b) {
 880         assert(b);
 881
 882         bus_reset_parsed_address(b);
 883         return bus_start_address(b);
 884 }
 885
 886 static int bus_start_fd(sd_bus *b) {
 887         struct stat st;
 888         int r;
 889
 890         assert(b);
 891         assert(b->input_fd >= 0);
 892         assert(b->output_fd >= 0);
 893
 894         r = fd_nonblock(b->input_fd, true);
 895         if (r < 0)
 896                 return r;
 897
 898         r = fd_cloexec(b->input_fd, true);
 899         if (r < 0)
 900                 return r;
 901
 902         if (b->input_fd != b->output_fd) {
 903                 r = fd_nonblock(b->output_fd, true);
 904                 if (r < 0)
 905                         return r;
 906
 907                 r = fd_cloexec(b->output_fd, true);
 908                 if (r < 0)
 909                         return r;
 910         }
 911
 912         if (fstat(b->input_fd, &st) < 0)
 913                 return -errno;
 914
 915         if (S_ISCHR(b->input_fd))
 916                 return bus_kernel_take_fd(b);
 917         else
 918                 return bus_socket_take_fd(b);
 919 }
 920
 921 int sd_bus_start(sd_bus *bus) {
 922         int r;
 923
 924         assert_return(bus, -EINVAL);
 925         assert_return(bus->state == BUS_UNSET, -EPERM);
 926         assert_return(!bus_pid_changed(bus), -ECHILD);
 927
 928         bus->state = BUS_OPENING;
 929
 930         if (bus->is_server && bus->bus_client)
 931                 return -EINVAL;
 932
 933         if (bus->input_fd >= 0)
 934                 r = bus_start_fd(bus);
 935         else if (bus->address || bus->sockaddr.sa.sa_family != AF_UNSPEC || bus->exec_path || bus->kernel)
 936                 r = bus_start_address(bus);
 937         else
 938                 return -EINVAL;
 939
 940         if (r < 0)
 941                 return r;
 942
 943         return bus_send_hello(bus);
 944 }
 945
 946 int sd_bus_open_system(sd_bus **ret) {
 947         const char *e;
 948         sd_bus *b;
 949         int r;
 950
 951         assert_return(ret, -EINVAL);
 952
 953         r = sd_bus_new(&b);
 954         if (r < 0)
 955                 return r;
 956
 957         e = secure_getenv("DBUS_SYSTEM_BUS_ADDRESS");
 958         if (e) {
 959                 r = sd_bus_set_address(b, e);
 960                 if (r < 0)
 961                         goto fail;
 962         } else {
 963                 b->sockaddr.un.sun_family = AF_UNIX;
 964                 strncpy(b->sockaddr.un.sun_path, "/run/dbus/system_bus_socket", sizeof(b->sockaddr.un.sun_path));
 965                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + sizeof("/run/dbus/system_bus_socket") - 1;
 966         }
 967
 968         b->bus_client = true;
 969
 970         r = sd_bus_start(b);
 971         if (r < 0)
 972                 goto fail;
 973
 974         *ret = b;
 975         return 0;
 976
 977 fail:
 978         bus_free(b);
 979         return r;
 980 }
 981
 982 int sd_bus_open_user(sd_bus **ret) {
 983         const char *e;
 984         sd_bus *b;
 985         size_t l;
 986         int r;
 987
 988         assert_return(ret, -EINVAL);
 989
 990         r = sd_bus_new(&b);
 991         if (r < 0)
 992                 return r;
 993
 994         e = secure_getenv("DBUS_SESSION_BUS_ADDRESS");
 995         if (e) {
 996                 r = sd_bus_set_address(b, e);
 997                 if (r < 0)
 998                         goto fail;
 999         } else {
1000                 e = secure_getenv("XDG_RUNTIME_DIR");
1001                 if (!e) {
1002                         r = -ENOENT;
1003                         goto fail;
1004                 }
1005
1006                 l = strlen(e);
1007                 if (l + 4 > sizeof(b->sockaddr.un.sun_path)) {
1008                         r = -E2BIG;
1009                         goto fail;
1010                 }
1011
1012                 b->sockaddr.un.sun_family = AF_UNIX;
1013                 memcpy(mempcpy(b->sockaddr.un.sun_path, e, l), "/bus", 4);
1014                 b->sockaddr_size = offsetof(struct sockaddr_un, sun_path) + l + 4;
1015         }
1016
1017         b->bus_client = true;
1018
1019         r = sd_bus_start(b);
1020         if (r < 0)
1021                 goto fail;
1022
1023         *ret = b;
1024         return 0;
1025
1026 fail:
1027         bus_free(b);
1028         return r;
1029 }
1030
1031 void sd_bus_close(sd_bus *bus) {
1032         if (!bus)
1033                 return;
1034         if (bus->state == BUS_CLOSED)
1035                 return;
1036         if (bus_pid_changed(bus))
1037                 return;
1038
1039         bus->state = BUS_CLOSED;
1040
1041         if (!bus->is_kernel)
1042                 bus_close_fds(bus);
1043
1044         /* We'll leave the fd open in case this is a kernel bus, since
1045          * there might still be memblocks around that reference this
1046          * bus, and they might need to invoke the
1047          * KDBUS_CMD_MSG_RELEASE ioctl on the fd when they are
1048          * freed. */
1049 }
1050
1051 sd_bus *sd_bus_ref(sd_bus *bus) {
1052         if (!bus)
1053                 return NULL;
1054
1055         assert_se(REFCNT_INC(bus->n_ref) >= 2);
1056
1057         return bus;
1058 }
1059
1060 sd_bus *sd_bus_unref(sd_bus *bus) {
1061         if (!bus)
1062                 return NULL;
1063
1064         if (REFCNT_DEC(bus->n_ref) <= 0)
1065                 bus_free(bus);
1066
1067         return NULL;
1068 }
1069
1070 int sd_bus_is_open(sd_bus *bus) {
1071
1072         assert_return(bus, -EINVAL);
1073         assert_return(!bus_pid_changed(bus), -ECHILD);
1074
1075         return BUS_IS_OPEN(bus->state);
1076 }
1077
1078 int sd_bus_can_send(sd_bus *bus, char type) {
1079         int r;
1080
1081         assert_return(bus, -EINVAL);
1082         assert_return(bus->state != BUS_UNSET, -ENOTCONN);
1083         assert_return(!bus_pid_changed(bus), -ECHILD);
1084
1085         if (type == SD_BUS_TYPE_UNIX_FD) {
1086                 if (!(bus->hello_flags & KDBUS_HELLO_ACCEPT_FD))
1087                         return 0;
1088
1089                 r = bus_ensure_running(bus);
1090                 if (r < 0)
1091                         return r;
1092
1093                 return bus->can_fds;
1094         }
1095
1096         return bus_type_is_valid(type);
1097 }
1098
1099 int sd_bus_get_server_id(sd_bus *bus, sd_id128_t *server_id) {
1100         int r;
1101
1102         assert_return(bus, -EINVAL);
1103         assert_return(server_id, -EINVAL);
1104         assert_return(!bus_pid_changed(bus), -ECHILD);
1105
1106         r = bus_ensure_running(bus);
1107         if (r < 0)
1108                 return r;
1109
1110         *server_id = bus->server_id;
1111         return 0;
1112 }
1113
1114 static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
1115         assert(m);
1116
1117         if (m->header->version > b->message_version)
1118                 return -EPERM;
1119
1120         if (m->sealed)
1121                 return 0;
1122
1123         return bus_message_seal(m, ++b->serial);
1124 }
1125
1126 static int dispatch_wqueue(sd_bus *bus) {
1127         int r, ret = 0;
1128
1129         assert(bus);
1130         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1131
1132         while (bus->wqueue_size > 0) {
1133
1134                 if (bus->is_kernel)
1135                         r = bus_kernel_write_message(bus, bus->wqueue[0]);
1136                 else
1137                         r = bus_socket_write_message(bus, bus->wqueue[0], &bus->windex);
1138
1139                 if (r < 0) {
1140                         sd_bus_close(bus);
1141                         return r;
1142                 } else if (r == 0)
1143                         /* Didn't do anything this time */
1144                         return ret;
1145                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
1146                         /* Fully written. Let's drop the entry from
1147                          * the queue.
1148                          *
1149                          * This isn't particularly optimized, but
1150                          * well, this is supposed to be our worst-case
1151                          * buffer only, and the socket buffer is
1152                          * supposed to be our primary buffer, and if
1153                          * it got full, then all bets are off
1154                          * anyway. */
1155
1156                         sd_bus_message_unref(bus->wqueue[0]);
1157                         bus->wqueue_size --;
1158                         memmove(bus->wqueue, bus->wqueue + 1, sizeof(sd_bus_message*) * bus->wqueue_size);
1159                         bus->windex = 0;
1160
1161                         ret = 1;
1162                 }
1163         }
1164
1165         return ret;
1166 }
1167
1168 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
1169         sd_bus_message *z = NULL;
1170         int r, ret = 0;
1171
1172         assert(bus);
1173         assert(m);
1174         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1175
1176         if (bus->rqueue_size > 0) {
1177                 /* Dispatch a queued message */
1178
1179                 *m = bus->rqueue[0];
1180                 bus->rqueue_size --;
1181                 memmove(bus->rqueue, bus->rqueue + 1, sizeof(sd_bus_message*) * bus->rqueue_size);
1182                 return 1;
1183         }
1184
1185         /* Try to read a new message */
1186         do {
1187                 if (bus->is_kernel)
1188                         r = bus_kernel_read_message(bus, &z);
1189                 else
1190                         r = bus_socket_read_message(bus, &z);
1191
1192                 if (r < 0) {
1193                         sd_bus_close(bus);
1194                         return r;
1195                 }
1196                 if (r == 0)
1197                         return ret;
1198
1199                 ret = 1;
1200         } while (!z);
1201
1202         *m = z;
1203         return ret;
1204 }
1205
1206 int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
1207         int r;
1208
1209         assert_return(bus, -EINVAL);
1210         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1211         assert_return(m, -EINVAL);
1212         assert_return(!bus_pid_changed(bus), -ECHILD);
1213
1214         if (m->n_fds > 0) {
1215                 r = sd_bus_can_send(bus, SD_BUS_TYPE_UNIX_FD);
1216                 if (r < 0)
1217                         return r;
1218                 if (r == 0)
1219                         return -ENOTSUP;
1220         }
1221
1222         /* If the serial number isn't kept, then we know that no reply
1223          * is expected */
1224         if (!serial && !m->sealed)
1225                 m->header->flags |= SD_BUS_MESSAGE_NO_REPLY_EXPECTED;
1226
1227         r = bus_seal_message(bus, m);
1228         if (r < 0)
1229                 return r;
1230
1231         /* If this is a reply and no reply was requested, then let's
1232          * suppress this, if we can */
1233         if (m->dont_send && !serial)
1234                 return 0;
1235
1236         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
1237                 size_t idx = 0;
1238
1239                 if (bus->is_kernel)
1240                         r = bus_kernel_write_message(bus, m);
1241                 else
1242                         r = bus_socket_write_message(bus, m, &idx);
1243
1244                 if (r < 0) {
1245                         sd_bus_close(bus);
1246                         return r;
1247                 } else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
1248                         /* Wasn't fully written. So let's remember how
1249                          * much was written. Note that the first entry
1250                          * of the wqueue array is always allocated so
1251                          * that we always can remember how much was
1252                          * written. */
1253                         bus->wqueue[0] = sd_bus_message_ref(m);
1254                         bus->wqueue_size = 1;
1255                         bus->windex = idx;
1256                 }
1257         } else {
1258                 sd_bus_message **q;
1259
1260                 /* Just append it to the queue. */
1261
1262                 if (bus->wqueue_size >= BUS_WQUEUE_MAX)
1263                         return -ENOBUFS;
1264
1265                 q = realloc(bus->wqueue, sizeof(sd_bus_message*) * (bus->wqueue_size + 1));
1266                 if (!q)
1267                         return -ENOMEM;
1268
1269                 bus->wqueue = q;
1270                 q[bus->wqueue_size ++] = sd_bus_message_ref(m);
1271         }
1272
1273         if (serial)
1274                 *serial = BUS_MESSAGE_SERIAL(m);
1275
1276         return 0;
1277 }
1278
1279 static usec_t calc_elapse(uint64_t usec) {
1280         if (usec == (uint64_t) -1)
1281                 return 0;
1282
1283         if (usec == 0)
1284                 usec = BUS_DEFAULT_TIMEOUT;
1285
1286         return now(CLOCK_MONOTONIC) + usec;
1287 }
1288
1289 static int timeout_compare(const void *a, const void *b) {
1290         const struct reply_callback *x = a, *y = b;
1291
1292         if (x->timeout != 0 && y->timeout == 0)
1293                 return -1;
1294
1295         if (x->timeout == 0 && y->timeout != 0)
1296                 return 1;
1297
1298         if (x->timeout < y->timeout)
1299                 return -1;
1300
1301         if (x->timeout > y->timeout)
1302                 return 1;
1303
1304         return 0;
1305 }
1306
1307 int sd_bus_send_with_reply(
1308                 sd_bus *bus,
1309                 sd_bus_message *m,
1310                 sd_bus_message_handler_t callback,
1311                 void *userdata,
1312                 uint64_t usec,
1313                 uint64_t *serial) {
1314
1315         struct reply_callback *c;
1316         int r;
1317
1318         assert_return(bus, -EINVAL);
1319         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1320         assert_return(m, -EINVAL);
1321         assert_return(m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL, -EINVAL);
1322         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1323         assert_return(callback, -EINVAL);
1324         assert_return(!bus_pid_changed(bus), -ECHILD);
1325
1326         r = hashmap_ensure_allocated(&bus->reply_callbacks, uint64_hash_func, uint64_compare_func);
1327         if (r < 0)
1328                 return r;
1329
1330         if (usec != (uint64_t) -1) {
1331                 r = prioq_ensure_allocated(&bus->reply_callbacks_prioq, timeout_compare);
1332                 if (r < 0)
1333                         return r;
1334         }
1335
1336         r = bus_seal_message(bus, m);
1337         if (r < 0)
1338                 return r;
1339
1340         c = new0(struct reply_callback, 1);
1341         if (!c)
1342                 return -ENOMEM;
1343
1344         c->callback = callback;
1345         c->userdata = userdata;
1346         c->serial = BUS_MESSAGE_SERIAL(m);
1347         c->timeout = calc_elapse(usec);
1348
1349         r = hashmap_put(bus->reply_callbacks, &c->serial, c);
1350         if (r < 0) {
1351                 free(c);
1352                 return r;
1353         }
1354
1355         if (c->timeout != 0) {
1356                 r = prioq_put(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1357                 if (r < 0) {
1358                         c->timeout = 0;
1359                         sd_bus_send_with_reply_cancel(bus, c->serial);
1360                         return r;
1361                 }
1362         }
1363
1364         r = sd_bus_send(bus, m, serial);
1365         if (r < 0) {
1366                 sd_bus_send_with_reply_cancel(bus, c->serial);
1367                 return r;
1368         }
1369
1370         return r;
1371 }
1372
1373 int sd_bus_send_with_reply_cancel(sd_bus *bus, uint64_t serial) {
1374         struct reply_callback *c;
1375
1376         assert_return(bus, -EINVAL);
1377         assert_return(serial != 0, -EINVAL);
1378         assert_return(!bus_pid_changed(bus), -ECHILD);
1379
1380         c = hashmap_remove(bus->reply_callbacks, &serial);
1381         if (!c)
1382                 return 0;
1383
1384         if (c->timeout != 0)
1385                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1386
1387         free(c);
1388         return 1;
1389 }
1390
1391 int bus_ensure_running(sd_bus *bus) {
1392         int r;
1393
1394         assert(bus);
1395
1396         if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED)
1397                 return -ENOTCONN;
1398         if (bus->state == BUS_RUNNING)
1399                 return 1;
1400
1401         for (;;) {
1402                 r = sd_bus_process(bus, NULL);
1403                 if (r < 0)
1404                         return r;
1405                 if (bus->state == BUS_RUNNING)
1406                         return 1;
1407                 if (r > 0)
1408                         continue;
1409
1410                 r = sd_bus_wait(bus, (uint64_t) -1);
1411                 if (r < 0)
1412                         return r;
1413         }
1414 }
1415
1416 int sd_bus_send_with_reply_and_block(
1417                 sd_bus *bus,
1418                 sd_bus_message *m,
1419                 uint64_t usec,
1420                 sd_bus_error *error,
1421                 sd_bus_message **reply) {
1422
1423         int r;
1424         usec_t timeout;
1425         uint64_t serial;
1426         bool room = false;
1427
1428         assert_return(bus, -EINVAL);
1429         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1430         assert_return(m, -EINVAL);
1431         assert_return(m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL, -EINVAL);
1432         assert_return(!(m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED), -EINVAL);
1433         assert_return(!bus_error_is_dirty(error), -EINVAL);
1434         assert_return(!bus_pid_changed(bus), -ECHILD);
1435
1436         r = bus_ensure_running(bus);
1437         if (r < 0)
1438                 return r;
1439
1440         r = sd_bus_send(bus, m, &serial);
1441         if (r < 0)
1442                 return r;
1443
1444         timeout = calc_elapse(usec);
1445
1446         for (;;) {
1447                 usec_t left;
1448                 sd_bus_message *incoming = NULL;
1449
1450                 if (!room) {
1451                         sd_bus_message **q;
1452
1453                         if (bus->rqueue_size >= BUS_RQUEUE_MAX)
1454                                 return -ENOBUFS;
1455
1456                         /* Make sure there's room for queuing this
1457                          * locally, before we read the message */
1458
1459                         q = realloc(bus->rqueue, (bus->rqueue_size + 1) * sizeof(sd_bus_message*));
1460                         if (!q)
1461                                 return -ENOMEM;
1462
1463                         bus->rqueue = q;
1464                         room = true;
1465                 }
1466
1467                 if (bus->is_kernel)
1468                         r = bus_kernel_read_message(bus, &incoming);
1469                 else
1470                         r = bus_socket_read_message(bus, &incoming);
1471                 if (r < 0)
1472                         return r;
1473                 if (incoming) {
1474
1475                         if (incoming->reply_serial == serial) {
1476                                 /* Found a match! */
1477
1478                                 if (incoming->header->type == SD_BUS_MESSAGE_TYPE_METHOD_RETURN) {
1479
1480                                         if (reply)
1481                                                 *reply = incoming;
1482                                         else
1483                                                 sd_bus_message_unref(incoming);
1484
1485                                         return 0;
1486                                 }
1487
1488                                 if (incoming->header->type == SD_BUS_MESSAGE_TYPE_METHOD_ERROR) {
1489                                         int k;
1490
1491                                         r = sd_bus_error_copy(error, &incoming->error);
1492                                         if (r < 0) {
1493                                                 sd_bus_message_unref(incoming);
1494                                                 return r;
1495                                         }
1496
1497                                         k = bus_error_to_errno(&incoming->error);
1498                                         sd_bus_message_unref(incoming);
1499                                         return k;
1500                                 }
1501
1502                                 sd_bus_message_unref(incoming);
1503                                 return -EIO;
1504                         }
1505
1506                         /* There's already guaranteed to be room for
1507                          * this, so need to resize things here */
1508                         bus->rqueue[bus->rqueue_size ++] = incoming;
1509                         room = false;
1510
1511                         /* Try to read more, right-away */
1512                         continue;
1513                 }
1514                 if (r != 0)
1515                         continue;
1516
1517                 if (timeout > 0) {
1518                         usec_t n;
1519
1520                         n = now(CLOCK_MONOTONIC);
1521                         if (n >= timeout)
1522                                 return -ETIMEDOUT;
1523
1524                         left = timeout - n;
1525                 } else
1526                         left = (uint64_t) -1;
1527
1528                 r = bus_poll(bus, true, left);
1529                 if (r < 0)
1530                         return r;
1531
1532                 r = dispatch_wqueue(bus);
1533                 if (r < 0)
1534                         return r;
1535         }
1536 }
1537
1538 int sd_bus_get_fd(sd_bus *bus) {
1539
1540         assert_return(bus, -EINVAL);
1541         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1542         assert_return(bus->input_fd == bus->output_fd, -EPERM);
1543         assert_return(!bus_pid_changed(bus), -ECHILD);
1544
1545         return bus->input_fd;
1546 }
1547
1548 int sd_bus_get_events(sd_bus *bus) {
1549         int flags = 0;
1550
1551         assert_return(bus, -EINVAL);
1552         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1553         assert_return(!bus_pid_changed(bus), -ECHILD);
1554
1555         if (bus->state == BUS_OPENING)
1556                 flags |= POLLOUT;
1557         else if (bus->state == BUS_AUTHENTICATING) {
1558
1559                 if (bus_socket_auth_needs_write(bus))
1560                         flags |= POLLOUT;
1561
1562                 flags |= POLLIN;
1563
1564         } else if (bus->state == BUS_RUNNING || bus->state == BUS_HELLO) {
1565                 if (bus->rqueue_size <= 0)
1566                         flags |= POLLIN;
1567                 if (bus->wqueue_size > 0)
1568                         flags |= POLLOUT;
1569         }
1570
1571         return flags;
1572 }
1573
1574 int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
1575         struct reply_callback *c;
1576
1577         assert_return(bus, -EINVAL);
1578         assert_return(timeout_usec, -EINVAL);
1579         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1580         assert_return(!bus_pid_changed(bus), -ECHILD);
1581
1582         if (bus->state == BUS_AUTHENTICATING) {
1583                 *timeout_usec = bus->auth_timeout;
1584                 return 1;
1585         }
1586
1587         if (bus->state != BUS_RUNNING && bus->state != BUS_HELLO) {
1588                 *timeout_usec = (uint64_t) -1;
1589                 return 0;
1590         }
1591
1592         c = prioq_peek(bus->reply_callbacks_prioq);
1593         if (!c) {
1594                 *timeout_usec = (uint64_t) -1;
1595                 return 0;
1596         }
1597
1598         *timeout_usec = c->timeout;
1599         return 1;
1600 }
1601
1602 static int process_timeout(sd_bus *bus) {
1603         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
1604         struct reply_callback *c;
1605         usec_t n;
1606         int r;
1607
1608         assert(bus);
1609
1610         c = prioq_peek(bus->reply_callbacks_prioq);
1611         if (!c)
1612                 return 0;
1613
1614         n = now(CLOCK_MONOTONIC);
1615         if (c->timeout > n)
1616                 return 0;
1617
1618         r = bus_message_new_synthetic_error(
1619                         bus,
1620                         c->serial,
1621                         &SD_BUS_ERROR_MAKE("org.freedesktop.DBus.Error.Timeout", "Timed out"),
1622                         &m);
1623         if (r < 0)
1624                 return r;
1625
1626         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
1627         hashmap_remove(bus->reply_callbacks, &c->serial);
1628
1629         r = c->callback(bus, m, c->userdata);
1630         free(c);
1631
1632         return r < 0 ? r : 1;
1633 }
1634
1635 static int process_hello(sd_bus *bus, sd_bus_message *m) {
1636         assert(bus);
1637         assert(m);
1638
1639         if (bus->state != BUS_HELLO)
1640                 return 0;
1641
1642         /* Let's make sure the first message on the bus is the HELLO
1643          * reply. But note that we don't actually parse the message
1644          * here (we leave that to the usual handling), we just verify
1645          * we don't let any earlier msg through. */
1646
1647         if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_RETURN &&
1648             m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_ERROR)
1649                 return -EIO;
1650
1651         if (m->reply_serial != bus->hello_serial)
1652                 return -EIO;
1653
1654         return 0;
1655 }
1656
1657 static int process_reply(sd_bus *bus, sd_bus_message *m) {
1658         struct reply_callback *c;
1659         int r;
1660
1661         assert(bus);
1662         assert(m);
1663
1664         if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_RETURN &&
1665             m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_ERROR)
1666                 return 0;
1667
1668         c = hashmap_remove(bus->reply_callbacks, &m->reply_serial);
1669         if (!c)
1670                 return 0;
1671
1672         if (c->timeout != 0)
1673                 prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
1674
1675         r = sd_bus_message_rewind(m, true);
1676         if (r < 0)
1677                 return r;
1678
1679         r = c->callback(bus, m, c->userdata);
1680         free(c);
1681
1682         return r;
1683 }
1684
1685 static int process_filter(sd_bus *bus, sd_bus_message *m) {
1686         struct filter_callback *l;
1687         int r;
1688
1689         assert(bus);
1690         assert(m);
1691
1692         do {
1693                 bus->filter_callbacks_modified = false;
1694
1695                 LIST_FOREACH(callbacks, l, bus->filter_callbacks) {
1696
1697                         if (bus->filter_callbacks_modified)
1698                                 break;
1699
1700                         /* Don't run this more than once per iteration */
1701                         if (l->last_iteration == bus->iteration_counter)
1702                                 continue;
1703
1704                         l->last_iteration = bus->iteration_counter;
1705
1706                         r = sd_bus_message_rewind(m, true);
1707                         if (r < 0)
1708                                 return r;
1709
1710                         r = l->callback(bus, m, l->userdata);
1711                         if (r != 0)
1712                                 return r;
1713
1714                 }
1715
1716         } while (bus->filter_callbacks_modified);
1717
1718         return 0;
1719 }
1720
1721 static int process_match(sd_bus *bus, sd_bus_message *m) {
1722         int r;
1723
1724         assert(bus);
1725         assert(m);
1726
1727         do {
1728                 bus->match_callbacks_modified = false;
1729
1730                 r = bus_match_run(bus, &bus->match_callbacks, m);
1731                 if (r != 0)
1732                         return r;
1733
1734         } while (bus->match_callbacks_modified);
1735
1736         return 0;
1737 }
1738
1739 static int process_builtin(sd_bus *bus, sd_bus_message *m) {
1740         _cleanup_bus_message_unref_ sd_bus_message *reply = NULL;
1741         int r;
1742
1743         assert(bus);
1744         assert(m);
1745
1746         if (m->header->type != SD_BUS_MESSAGE_TYPE_METHOD_CALL)
1747                 return 0;
1748
1749         if (!streq_ptr(m->interface, "org.freedesktop.DBus.Peer"))
1750                 return 0;
1751
1752         if (m->header->flags & SD_BUS_MESSAGE_NO_REPLY_EXPECTED)
1753                 return 1;
1754
1755         if (streq_ptr(m->member, "Ping"))
1756                 r = sd_bus_message_new_method_return(bus, m, &reply);
1757         else if (streq_ptr(m->member, "GetMachineId")) {
1758                 sd_id128_t id;
1759                 char sid[33];
1760
1761                 r = sd_id128_get_machine(&id);
1762                 if (r < 0)
1763                         return r;
1764
1765                 r = sd_bus_message_new_method_return(bus, m, &reply);
1766                 if (r < 0)
1767                         return r;
1768
1769                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
1770         } else {
1771                 r = sd_bus_message_new_method_errorf(
1772                                 bus, m, &reply,
1773                                 "org.freedesktop.DBus.Error.UnknownMethod",
1774                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
1775         }
1776
1777         if (r < 0)
1778                 return r;
1779
1780         r = sd_bus_send(bus, reply, NULL);
1781         if (r < 0)
1782                 return r;
1783
1784         return 1;
1785 }
1786
1787 static int process_message(sd_bus *bus, sd_bus_message *m) {
1788         int r;
1789
1790         assert(bus);
1791         assert(m);
1792
1793         bus->iteration_counter++;
1794
1795         r = process_hello(bus, m);
1796         if (r != 0)
1797                 return r;
1798
1799         r = process_reply(bus, m);
1800         if (r != 0)
1801                 return r;
1802
1803         r = process_filter(bus, m);
1804         if (r != 0)
1805                 return r;
1806
1807         r = process_match(bus, m);
1808         if (r != 0)
1809                 return r;
1810
1811         r = process_builtin(bus, m);
1812         if (r != 0)
1813                 return r;
1814
1815         return bus_process_object(bus, m);
1816 }
1817
1818 static int process_running(sd_bus *bus, sd_bus_message **ret) {
1819         _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
1820         int r;
1821
1822         assert(bus);
1823         assert(bus->state == BUS_RUNNING || bus->state == BUS_HELLO);
1824
1825         r = process_timeout(bus);
1826         if (r != 0)
1827                 goto null_message;
1828
1829         r = dispatch_wqueue(bus);
1830         if (r != 0)
1831                 goto null_message;
1832
1833         r = dispatch_rqueue(bus, &m);
1834         if (r < 0)
1835                 return r;
1836         if (!m)
1837                 goto null_message;
1838
1839         r = process_message(bus, m);
1840         if (r != 0)
1841                 goto null_message;
1842
1843         if (ret) {
1844                 r = sd_bus_message_rewind(m, true);
1845                 if (r < 0)
1846                         return r;
1847
1848                 *ret = m;
1849                 m = NULL;
1850                 return 1;
1851         }
1852
1853         if (m->header->type == SD_BUS_MESSAGE_TYPE_METHOD_CALL) {
1854
1855                 r = sd_bus_reply_method_errorf(
1856                                 bus, m,
1857                                 "org.freedesktop.DBus.Error.UnknownObject",
1858                                 "Unknown object '%s'.", m->path);
1859                 if (r < 0)
1860                         return r;
1861         }
1862
1863         return 1;
1864
1865 null_message:
1866         if (r >= 0 && ret)
1867                 *ret = NULL;
1868
1869         return r;
1870 }
1871
1872 int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
1873         int r;
1874
1875         /* Returns 0 when we didn't do anything. This should cause the
1876          * caller to invoke sd_bus_wait() before returning the next
1877          * time. Returns > 0 when we did something, which possibly
1878          * means *ret is filled in with an unprocessed message. */
1879
1880         assert_return(bus, -EINVAL);
1881         assert_return(!bus_pid_changed(bus), -ECHILD);
1882
1883         /* We don't allow recursively invoking sd_bus_process(). */
1884         assert_return(!bus->processing, -EBUSY);
1885
1886         switch (bus->state) {
1887
1888         case BUS_UNSET:
1889         case BUS_CLOSED:
1890                 return -ENOTCONN;
1891
1892         case BUS_OPENING:
1893                 r = bus_socket_process_opening(bus);
1894                 if (r < 0)
1895                         return r;
1896                 if (ret)
1897                         *ret = NULL;
1898                 return r;
1899
1900         case BUS_AUTHENTICATING:
1901
1902                 r = bus_socket_process_authenticating(bus);
1903                 if (r < 0)
1904                         return r;
1905                 if (ret)
1906                         *ret = NULL;
1907                 return r;
1908
1909         case BUS_RUNNING:
1910         case BUS_HELLO:
1911
1912                 bus->processing = true;
1913                 r = process_running(bus, ret);
1914                 bus->processing = false;
1915
1916                 return r;
1917         }
1918
1919         assert_not_reached("Unknown state");
1920 }
1921
1922 static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
1923         struct pollfd p[2] = {};
1924         int r, e, n;
1925         struct timespec ts;
1926         usec_t until, m;
1927
1928         assert(bus);
1929         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1930
1931         e = sd_bus_get_events(bus);
1932         if (e < 0)
1933                 return e;
1934
1935         if (need_more)
1936                 e |= POLLIN;
1937
1938         r = sd_bus_get_timeout(bus, &until);
1939         if (r < 0)
1940                 return r;
1941         if (r == 0)
1942                 m = (uint64_t) -1;
1943         else {
1944                 usec_t nw;
1945                 nw = now(CLOCK_MONOTONIC);
1946                 m = until > nw ? until - nw : 0;
1947         }
1948
1949         if (timeout_usec != (uint64_t) -1 && (m == (uint64_t) -1 || timeout_usec < m))
1950                 m = timeout_usec;
1951
1952         p[0].fd = bus->input_fd;
1953         if (bus->output_fd == bus->input_fd) {
1954                 p[0].events = e;
1955                 n = 1;
1956         } else {
1957                 p[0].events = e & POLLIN;
1958                 p[1].fd = bus->output_fd;
1959                 p[1].events = e & POLLOUT;
1960                 n = 2;
1961         }
1962
1963         r = ppoll(p, n, m == (uint64_t) -1 ? NULL : timespec_store(&ts, m), NULL);
1964         if (r < 0)
1965                 return -errno;
1966
1967         return r > 0 ? 1 : 0;
1968 }
1969
1970 int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
1971
1972         assert_return(bus, -EINVAL);
1973         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1974         assert_return(!bus_pid_changed(bus), -ECHILD);
1975
1976         if (bus->rqueue_size > 0)
1977                 return 0;
1978
1979         return bus_poll(bus, false, timeout_usec);
1980 }
1981
1982 int sd_bus_flush(sd_bus *bus) {
1983         int r;
1984
1985         assert_return(bus, -EINVAL);
1986         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
1987         assert_return(!bus_pid_changed(bus), -ECHILD);
1988
1989         r = bus_ensure_running(bus);
1990         if (r < 0)
1991                 return r;
1992
1993         if (bus->wqueue_size <= 0)
1994                 return 0;
1995
1996         for (;;) {
1997                 r = dispatch_wqueue(bus);
1998                 if (r < 0)
1999                         return r;
2000
2001                 if (bus->wqueue_size <= 0)
2002                         return 0;
2003
2004                 r = bus_poll(bus, false, (uint64_t) -1);
2005                 if (r < 0)
2006                         return r;
2007         }
2008 }
2009
2010 int sd_bus_add_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
2011         struct filter_callback *f;
2012
2013         assert_return(bus, -EINVAL);
2014         assert_return(callback, -EINVAL);
2015         assert_return(!bus_pid_changed(bus), -ECHILD);
2016
2017         f = new0(struct filter_callback, 1);
2018         if (!f)
2019                 return -ENOMEM;
2020         f->callback = callback;
2021         f->userdata = userdata;
2022
2023         bus->filter_callbacks_modified = true;
2024         LIST_PREPEND(callbacks, bus->filter_callbacks, f);
2025         return 0;
2026 }
2027
2028 int sd_bus_remove_filter(sd_bus *bus, sd_bus_message_handler_t callback, void *userdata) {
2029         struct filter_callback *f;
2030
2031         assert_return(bus, -EINVAL);
2032         assert_return(callback, -EINVAL);
2033         assert_return(!bus_pid_changed(bus), -ECHILD);
2034
2035         LIST_FOREACH(callbacks, f, bus->filter_callbacks) {
2036                 if (f->callback == callback && f->userdata == userdata) {
2037                         bus->filter_callbacks_modified = true;
2038                         LIST_REMOVE(callbacks, bus->filter_callbacks, f);
2039                         free(f);
2040                         return 1;
2041                 }
2042         }
2043
2044         return 0;
2045 }
2046
2047 int sd_bus_add_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
2048         struct bus_match_component *components = NULL;
2049         unsigned n_components = 0;
2050         uint64_t cookie = 0;
2051         int r = 0;
2052
2053         assert_return(bus, -EINVAL);
2054         assert_return(match, -EINVAL);
2055         assert_return(!bus_pid_changed(bus), -ECHILD);
2056
2057         r = bus_match_parse(match, &components, &n_components);
2058         if (r < 0)
2059                 goto finish;
2060
2061         if (bus->bus_client) {
2062                 cookie = ++bus->match_cookie;
2063
2064                 r = bus_add_match_internal(bus, match, components, n_components, cookie);
2065                 if (r < 0)
2066                         goto finish;
2067         }
2068
2069         bus->match_callbacks_modified = true;
2070         r = bus_match_add(&bus->match_callbacks, components, n_components, callback, userdata, cookie, NULL);
2071         if (r < 0) {
2072                 if (bus->bus_client)
2073                         bus_remove_match_internal(bus, match, cookie);
2074         }
2075
2076 finish:
2077         bus_match_parse_free(components, n_components);
2078         return r;
2079 }
2080
2081 int sd_bus_remove_match(sd_bus *bus, const char *match, sd_bus_message_handler_t callback, void *userdata) {
2082         struct bus_match_component *components = NULL;
2083         unsigned n_components = 0;
2084         int r = 0, q = 0;
2085         uint64_t cookie = 0;
2086
2087         assert_return(bus, -EINVAL);
2088         assert_return(match, -EINVAL);
2089         assert_return(!bus_pid_changed(bus), -ECHILD);
2090
2091         r = bus_match_parse(match, &components, &n_components);
2092         if (r < 0)
2093                 return r;
2094
2095         bus->match_callbacks_modified = true;
2096         r = bus_match_remove(&bus->match_callbacks, components, n_components, callback, userdata, &cookie);
2097
2098         if (bus->bus_client)
2099                 q = bus_remove_match_internal(bus, match, cookie);
2100
2101         bus_match_parse_free(components, n_components);
2102
2103         return r < 0 ? r : q;
2104 }
2105
2106 bool bus_pid_changed(sd_bus *bus) {
2107         assert(bus);
2108
2109         /* We don't support people creating a bus connection and
2110          * keeping it around over a fork(). Let's complain. */
2111
2112         return bus->original_pid != getpid();
2113 }