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