X-Git-Url: https://www.chiark.greenend.org.uk/ucgi/~ianmdlvl/git?p=elogind.git;a=blobdiff_plain;f=src%2Flibsystemd-bus%2Fsd-bus.c;h=f13e346dcfb105f0830e60e957b94e838b467bde;hp=fdfbbeb2be0a30933f1531586e3d394debe40e50;hb=c99c09a2dc0f72e6f10449c9e110158612a43ab6;hpb=c49b30a23583ff39daaa26696bcab478d2fee0bb

diff --git a/src/libsystemd-bus/sd-bus.c b/src/libsystemd-bus/sd-bus.c
index fdfbbeb2b..f13e346dc 100644
--- a/src/libsystemd-bus/sd-bus.c
+++ b/src/libsystemd-bus/sd-bus.c
@@ -99,10 +99,26 @@ static void bus_node_destroy(sd_bus *b, struct node *n) {
         free(n);
 }
 
+static void bus_reset_queues(sd_bus *b) {
+        unsigned i;
+
+        assert(b);
+
+        for (i = 0; i < b->rqueue_size; i++)
+                sd_bus_message_unref(b->rqueue[i]);
+        free(b->rqueue);
+
+        for (i = 0; i < b->wqueue_size; i++)
+                sd_bus_message_unref(b->wqueue[i]);
+        free(b->wqueue);
+
+        b->rqueue = b->wqueue = NULL;
+        b->rqueue_size = b->wqueue_size = 0;
+}
+
 static void bus_free(sd_bus *b) {
         struct filter_callback *f;
         struct node *n;
-        unsigned i;
 
         assert(b);
 
@@ -126,13 +142,7 @@ static void bus_free(sd_bus *b) {
         close_many(b->fds, b->n_fds);
         free(b->fds);
 
-        for (i = 0; i < b->rqueue_size; i++)
-                sd_bus_message_unref(b->rqueue[i]);
-        free(b->rqueue);
-
-        for (i = 0; i < b->wqueue_size; i++)
-                sd_bus_message_unref(b->wqueue[i]);
-        free(b->wqueue);
+        bus_reset_queues(b);
 
         hashmap_free_free(b->reply_callbacks);
         prioq_free(b->reply_callbacks_prioq);
@@ -264,6 +274,24 @@ _public_ int sd_bus_negotiate_fds(sd_bus *bus, int b) {
         return 0;
 }
 
+_public_ int sd_bus_negotiate_attach_timestamp(sd_bus *bus, int b) {
+        assert_return(bus, -EINVAL);
+        assert_return(bus->state == BUS_UNSET, -EPERM);
+        assert_return(!bus_pid_changed(bus), -ECHILD);
+
+        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_TIMESTAMP, b);
+        return 0;
+}
+
+_public_ int sd_bus_negotiate_attach_creds(sd_bus *bus, int b) {
+        assert_return(bus, -EINVAL);
+        assert_return(bus->state == BUS_UNSET, -EPERM);
+        assert_return(!bus_pid_changed(bus), -ECHILD);
+
+        SET_FLAG(bus->hello_flags, KDBUS_HELLO_ATTACH_CREDS, b);
+        return 0;
+}
+
 _public_ int sd_bus_negotiate_attach_comm(sd_bus *bus, int b) {
         assert_return(bus, -EINVAL);
         assert_return(bus->state == BUS_UNSET, -EPERM);
@@ -347,12 +375,12 @@ _public_ int sd_bus_set_anonymous(sd_bus *bus, int b) {
         return 0;
 }
 
-static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
+static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata, sd_bus_error *error) {
         const char *s;
         int r;
 
         assert(bus);
-        assert(bus->state == BUS_HELLO);
+        assert(bus->state == BUS_HELLO || bus->state == BUS_CLOSING);
         assert(reply);
 
         r = sd_bus_message_get_errno(reply);
@@ -372,7 +400,8 @@ static int hello_callback(sd_bus *bus, sd_bus_message *reply, void *userdata) {
         if (!bus->unique_name)
                 return -ENOMEM;
 
-        bus->state = BUS_RUNNING;
+        if (bus->state == BUS_HELLO)
+                bus->state = BUS_RUNNING;
 
         return 1;
 }
@@ -1165,6 +1194,7 @@ _public_ int sd_bus_open_system_container(const char *machine, sd_bus **ret) {
 }
 
 _public_ void sd_bus_close(sd_bus *bus) {
+
         if (!bus)
                 return;
         if (bus->state == BUS_CLOSED)
@@ -1176,6 +1206,10 @@ _public_ void sd_bus_close(sd_bus *bus) {
 
         sd_bus_detach_event(bus);
 
+        /* Drop all queued messages so that they drop references to
+         * the bus object and the bus may be freed */
+        bus_reset_queues(bus);
+
         if (!bus->is_kernel)
                 bus_close_fds(bus);
 
@@ -1186,6 +1220,18 @@ _public_ void sd_bus_close(sd_bus *bus) {
          * freed. */
 }
 
+static void bus_enter_closing(sd_bus *bus) {
+        assert(bus);
+
+        if (bus->state != BUS_OPENING &&
+            bus->state != BUS_AUTHENTICATING &&
+            bus->state != BUS_HELLO &&
+            bus->state != BUS_RUNNING)
+                return;
+
+        bus->state = BUS_CLOSING;
+}
+
 _public_ sd_bus *sd_bus_ref(sd_bus *bus) {
         assert_return(bus, NULL);
 
@@ -1253,12 +1299,31 @@ static int bus_seal_message(sd_bus *b, sd_bus_message *m) {
         if (m->header->version > b->message_version)
                 return -EPERM;
 
-        if (m->sealed)
+        if (m->sealed) {
+                /* If we copy the same message to multiple
+                 * destinations, avoid using the same serial
+                 * numbers. */
+                b->serial = MAX(b->serial, BUS_MESSAGE_SERIAL(m));
                 return 0;
+        }
 
         return bus_message_seal(m, ++b->serial);
 }
 
+static int bus_write_message(sd_bus *bus, sd_bus_message *message, size_t *idx) {
+        int r;
+
+        assert(bus);
+        assert(message);
+
+        if (bus->is_kernel)
+                r = bus_kernel_write_message(bus, message);
+        else
+                r = bus_socket_write_message(bus, message, idx);
+
+        return r;
+}
+
 static int dispatch_wqueue(sd_bus *bus) {
         int r, ret = 0;
 
@@ -1267,15 +1332,10 @@ static int dispatch_wqueue(sd_bus *bus) {
 
         while (bus->wqueue_size > 0) {
 
-                if (bus->is_kernel)
-                        r = bus_kernel_write_message(bus, bus->wqueue[0]);
-                else
-                        r = bus_socket_write_message(bus, bus->wqueue[0], &bus->windex);
-
-                if (r < 0) {
-                        sd_bus_close(bus);
+                r = bus_write_message(bus, bus->wqueue[0], &bus->windex);
+                if (r < 0)
                         return r;
-                } else if (r == 0)
+                else if (r == 0)
                         /* Didn't do anything this time */
                         return ret;
                 else if (bus->is_kernel || bus->windex >= BUS_MESSAGE_SIZE(bus->wqueue[0])) {
@@ -1301,6 +1361,20 @@ static int dispatch_wqueue(sd_bus *bus) {
         return ret;
 }
 
+static int bus_read_message(sd_bus *bus, sd_bus_message **m) {
+        int r;
+
+        assert(bus);
+        assert(m);
+
+        if (bus->is_kernel)
+                r = bus_kernel_read_message(bus, m);
+        else
+                r = bus_socket_read_message(bus, m);
+
+        return r;
+}
+
 static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
         sd_bus_message *z = NULL;
         int r, ret = 0;
@@ -1320,15 +1394,9 @@ static int dispatch_rqueue(sd_bus *bus, sd_bus_message **m) {
 
         /* Try to read a new message */
         do {
-                if (bus->is_kernel)
-                        r = bus_kernel_read_message(bus, &z);
-                else
-                        r = bus_socket_read_message(bus, &z);
-
-                if (r < 0) {
-                        sd_bus_close(bus);
+                r = bus_read_message(bus, &z);
+                if (r < 0)
                         return r;
-                }
                 if (r == 0)
                         return ret;
 
@@ -1372,15 +1440,10 @@ _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         if ((bus->state == BUS_RUNNING || bus->state == BUS_HELLO) && bus->wqueue_size <= 0) {
                 size_t idx = 0;
 
-                if (bus->is_kernel)
-                        r = bus_kernel_write_message(bus, m);
-                else
-                        r = bus_socket_write_message(bus, m, &idx);
-
-                if (r < 0) {
-                        sd_bus_close(bus);
+                r = bus_write_message(bus, m, &idx);
+                if (r < 0)
                         return r;
-                } else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
+                else if (!bus->is_kernel && idx < BUS_MESSAGE_SIZE(m))  {
                         /* Wasn't fully written. So let's remember how
                          * much was written. Note that the first entry
                          * of the wqueue array is always allocated so
@@ -1412,6 +1475,27 @@ _public_ int sd_bus_send(sd_bus *bus, sd_bus_message *m, uint64_t *serial) {
         return 1;
 }
 
+_public_ int sd_bus_send_to(sd_bus *bus, sd_bus_message *m, const char *destination, uint64_t *serial) {
+        int r;
+
+        assert_return(bus, -EINVAL);
+        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(m, -EINVAL);
+        assert_return(!bus_pid_changed(bus), -ECHILD);
+
+        if (!streq_ptr(m->destination, destination)) {
+
+                if (!destination)
+                        return -EEXIST;
+
+                r = sd_bus_message_set_destination(m, destination);
+                if (r < 0)
+                        return r;
+        }
+
+        return sd_bus_send(bus, m, serial);
+}
+
 static usec_t calc_elapse(uint64_t usec) {
         if (usec == (uint64_t) -1)
                 return 0;
@@ -1529,7 +1613,7 @@ int bus_ensure_running(sd_bus *bus) {
 
         assert(bus);
 
-        if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED)
+        if (bus->state == BUS_UNSET || bus->state == BUS_CLOSED || bus->state == BUS_CLOSING)
                 return -ENOTCONN;
         if (bus->state == BUS_RUNNING)
                 return 1;
@@ -1600,12 +1684,10 @@ _public_ int sd_bus_call(
                         room = true;
                 }
 
-                if (bus->is_kernel)
-                        r = bus_kernel_read_message(bus, &incoming);
-                else
-                        r = bus_socket_read_message(bus, &incoming);
+                r = bus_read_message(bus, &incoming);
                 if (r < 0)
                         return r;
+
                 if (incoming) {
 
                         if (incoming->reply_serial == serial) {
@@ -1637,6 +1719,19 @@ _public_ int sd_bus_call(
 
                                 sd_bus_message_unref(incoming);
                                 return -EIO;
+
+                        } else if (incoming->header->serial == serial &&
+                                   bus->unique_name &&
+                                   incoming->sender &&
+                                   streq(bus->unique_name, incoming->sender)) {
+
+                                /* Our own message? Somebody is trying
+                                 * to send its own client a message,
+                                 * let's not dead-lock, let's fail
+                                 * immediately. */
+
+                                sd_bus_message_unref(incoming);
+                                return -ELOOP;
                         }
 
                         /* There's already guaranteed to be room for
@@ -1674,7 +1769,6 @@ _public_ int sd_bus_call(
 _public_ int sd_bus_get_fd(sd_bus *bus) {
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(bus->input_fd == bus->output_fd, -EPERM);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
@@ -1685,7 +1779,7 @@ _public_ int sd_bus_get_events(sd_bus *bus) {
         int flags = 0;
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
         if (bus->state == BUS_OPENING)
@@ -1712,9 +1806,14 @@ _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
 
         assert_return(bus, -EINVAL);
         assert_return(timeout_usec, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+        assert_return(BUS_IS_OPEN(bus->state) || bus->state == BUS_CLOSING, -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING) {
+                *timeout_usec = 0;
+                return 1;
+        }
+
         if (bus->state == BUS_AUTHENTICATING) {
                 *timeout_usec = bus->auth_timeout;
                 return 1;
@@ -1741,6 +1840,7 @@ _public_ int sd_bus_get_timeout(sd_bus *bus, uint64_t *timeout_usec) {
 }
 
 static int process_timeout(sd_bus *bus) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         _cleanup_bus_message_unref_ sd_bus_message* m = NULL;
         struct reply_callback *c;
         usec_t n;
@@ -1759,18 +1859,28 @@ static int process_timeout(sd_bus *bus) {
         r = bus_message_new_synthetic_error(
                         bus,
                         c->serial,
-                        &SD_BUS_ERROR_MAKE(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
+                        &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Method call timed out"),
                         &m);
         if (r < 0)
                 return r;
 
+        r = bus_seal_message(bus, m);
+        if (r < 0)
+                return r;
+
         assert_se(prioq_pop(bus->reply_callbacks_prioq) == c);
         hashmap_remove(bus->reply_callbacks, &c->serial);
 
-        r = c->callback(bus, m, c->userdata);
+        bus->current = m;
+        bus->iteration_counter ++;
+
+        r = c->callback(bus, m, c->userdata, &error_buffer);
+        r = bus_maybe_reply_error(m, r, &error_buffer);
         free(c);
 
-        return r < 0 ? r : 1;
+        bus->current = NULL;
+
+        return r;
 }
 
 static int process_hello(sd_bus *bus, sd_bus_message *m) {
@@ -1796,6 +1906,7 @@ static int process_hello(sd_bus *bus, sd_bus_message *m) {
 }
 
 static int process_reply(sd_bus *bus, sd_bus_message *m) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         struct reply_callback *c;
         int r;
 
@@ -1817,13 +1928,15 @@ static int process_reply(sd_bus *bus, sd_bus_message *m) {
         if (r < 0)
                 return r;
 
-        r = c->callback(bus, m, c->userdata);
+        r = c->callback(bus, m, c->userdata, &error_buffer);
+        r = bus_maybe_reply_error(m, r, &error_buffer);
         free(c);
 
         return r;
 }
 
 static int process_filter(sd_bus *bus, sd_bus_message *m) {
+        _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
         struct filter_callback *l;
         int r;
 
@@ -1848,7 +1961,8 @@ static int process_filter(sd_bus *bus, sd_bus_message *m) {
                         if (r < 0)
                                 return r;
 
-                        r = l->callback(bus, m, l->userdata);
+                        r = l->callback(bus, m, l->userdata, &error_buffer);
+                        r = bus_maybe_reply_error(m, r, &error_buffer);
                         if (r != 0)
                                 return r;
 
@@ -1894,7 +2008,7 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
                 return 1;
 
         if (streq_ptr(m->member, "Ping"))
-                r = sd_bus_message_new_method_return(bus, m, &reply);
+                r = sd_bus_message_new_method_return(m, &reply);
         else if (streq_ptr(m->member, "GetMachineId")) {
                 sd_id128_t id;
                 char sid[33];
@@ -1903,14 +2017,14 @@ static int process_builtin(sd_bus *bus, sd_bus_message *m) {
                 if (r < 0)
                         return r;
 
-                r = sd_bus_message_new_method_return(bus, m, &reply);
+                r = sd_bus_message_new_method_return(m, &reply);
                 if (r < 0)
                         return r;
 
                 r = sd_bus_message_append(reply, "s", sd_id128_to_string(id, sid));
         } else {
                 r = sd_bus_message_new_method_errorf(
-                                bus, m, &reply,
+                                m, &reply,
                                 SD_BUS_ERROR_UNKNOWN_METHOD,
                                  "Unknown method '%s' on interface '%s'.", m->member, m->interface);
         }
@@ -2005,7 +2119,7 @@ static int process_running(sd_bus *bus, sd_bus_message **ret) {
         if (m->header->type == SD_BUS_MESSAGE_METHOD_CALL) {
 
                 r = sd_bus_reply_method_errorf(
-                                bus, m,
+                                m,
                                 SD_BUS_ERROR_UNKNOWN_OBJECT,
                                 "Unknown object '%s'.", m->path);
                 if (r < 0)
@@ -2021,6 +2135,85 @@ null_message:
         return r;
 }
 
+static int process_closing(sd_bus *bus, sd_bus_message **ret) {
+        _cleanup_bus_message_unref_ sd_bus_message *m = NULL;
+        struct reply_callback *c;
+        int r;
+
+        assert(bus);
+        assert(bus->state == BUS_CLOSING);
+
+        c = hashmap_first(bus->reply_callbacks);
+        if (c) {
+                _cleanup_bus_error_free_ sd_bus_error error_buffer = SD_BUS_ERROR_NULL;
+
+                /* First, fail all outstanding method calls */
+                r = bus_message_new_synthetic_error(
+                                bus,
+                                c->serial,
+                                &SD_BUS_ERROR_MAKE_CONST(SD_BUS_ERROR_NO_REPLY, "Connection terminated"),
+                                &m);
+                if (r < 0)
+                        return r;
+
+                r = bus_seal_message(bus, m);
+                if (r < 0)
+                        return r;
+
+                if (c->timeout != 0)
+                        prioq_remove(bus->reply_callbacks_prioq, c, &c->prioq_idx);
+
+                hashmap_remove(bus->reply_callbacks, &c->serial);
+
+                bus->current = m;
+                bus->iteration_counter++;
+
+                r = c->callback(bus, m, c->userdata, &error_buffer);
+                r = bus_maybe_reply_error(m, r, &error_buffer);
+                free(c);
+
+                goto finish;
+        }
+
+        /* Then, synthesize a Disconnected message */
+        r = sd_bus_message_new_signal(
+                        bus,
+                        "/org/freedesktop/DBus/Local",
+                        "org.freedesktop.DBus.Local",
+                        "Disconnected",
+                        &m);
+        if (r < 0)
+                return r;
+
+        r = bus_seal_message(bus, m);
+        if (r < 0)
+                return r;
+
+        sd_bus_close(bus);
+
+        bus->current = m;
+        bus->iteration_counter++;
+
+        r = process_filter(bus, m);
+        if (r != 0)
+                goto finish;
+
+        r = process_match(bus, m);
+        if (r != 0)
+                goto finish;
+
+        if (ret) {
+                *ret = m;
+                m = NULL;
+        }
+
+        r = 1;
+
+finish:
+        bus->current = NULL;
+        return r;
+}
+
 _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
         BUS_DONT_DESTROY(bus);
         int r;
@@ -2034,7 +2227,7 @@ _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
         /* We don't allow recursively invoking sd_bus_process(). */
-        assert_return(!bus->processing, -EBUSY);
+        assert_return(!bus->current, -EBUSY);
 
         switch (bus->state) {
 
@@ -2044,29 +2237,43 @@ _public_ int sd_bus_process(sd_bus *bus, sd_bus_message **ret) {
 
         case BUS_OPENING:
                 r = bus_socket_process_opening(bus);
-                if (r < 0)
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+                } else if (r < 0)
                         return r;
                 if (ret)
                         *ret = NULL;
                 return r;
 
         case BUS_AUTHENTICATING:
-
                 r = bus_socket_process_authenticating(bus);
-                if (r < 0)
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+                } else if (r < 0)
                         return r;
+
                 if (ret)
                         *ret = NULL;
+
                 return r;
 
         case BUS_RUNNING:
         case BUS_HELLO:
-
-                bus->processing = true;
                 r = process_running(bus, ret);
-                bus->processing = false;
+                if (r == -ECONNRESET || r == -EPIPE) {
+                        bus_enter_closing(bus);
+                        r = 1;
+
+                        if (ret)
+                                *ret = NULL;
+                }
 
                 return r;
+
+        case BUS_CLOSING:
+                return process_closing(bus, ret);
         }
 
         assert_not_reached("Unknown state");
@@ -2079,6 +2286,10 @@ static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
         usec_t m = (usec_t) -1;
 
         assert(bus);
+
+        if (bus->state == BUS_CLOSING)
+                return 1;
+
         assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
 
         e = sd_bus_get_events(bus);
@@ -2129,9 +2340,13 @@ static int bus_poll(sd_bus *bus, bool need_more, uint64_t timeout_usec) {
 _public_ int sd_bus_wait(sd_bus *bus, uint64_t timeout_usec) {
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING)
+                return 0;
+
+        assert_return(BUS_IS_OPEN(bus->state) , -ENOTCONN);
+
         if (bus->rqueue_size > 0)
                 return 0;
 
@@ -2142,9 +2357,13 @@ _public_ int sd_bus_flush(sd_bus *bus) {
         int r;
 
         assert_return(bus, -EINVAL);
-        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
         assert_return(!bus_pid_changed(bus), -ECHILD);
 
+        if (bus->state == BUS_CLOSING)
+                return 0;
+
+        assert_return(BUS_IS_OPEN(bus->state), -ENOTCONN);
+
         r = bus_ensure_running(bus);
         if (r < 0)
                 return r;
@@ -2286,7 +2505,7 @@ bool bus_pid_changed(sd_bus *bus) {
 }
 
 static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
-        void *bus = userdata;
+        sd_bus *bus = userdata;
         int r;
 
         assert(bus);
@@ -2299,7 +2518,7 @@ static int io_callback(sd_event_source *s, int fd, uint32_t revents, void *userd
 }
 
 static int time_callback(sd_event_source *s, uint64_t usec, void *userdata) {
-        void *bus = userdata;
+        sd_bus *bus = userdata;
         int r;
 
         assert(bus);
@@ -2370,16 +2589,21 @@ _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
         int r;
 
         assert_return(bus, -EINVAL);
-        assert_return(event, -EINVAL);
         assert_return(!bus->event, -EBUSY);
 
         assert(!bus->input_io_event_source);
         assert(!bus->output_io_event_source);
         assert(!bus->time_event_source);
 
-        bus->event = sd_event_ref(event);
+        if (event)
+                bus->event = sd_event_ref(event);
+        else  {
+                r = sd_event_default(&bus->event);
+                if (r < 0)
+                        return r;
+        }
 
-        r = sd_event_add_io(event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
+        r = sd_event_add_io(bus->event, bus->input_fd, 0, io_callback, bus, &bus->input_io_event_source);
         if (r < 0)
                 goto fail;
 
@@ -2388,7 +2612,7 @@ _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
                 goto fail;
 
         if (bus->output_fd != bus->input_fd) {
-                r = sd_event_add_io(event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
+                r = sd_event_add_io(bus->event, bus->output_fd, 0, io_callback, bus, &bus->output_io_event_source);
                 if (r < 0)
                         goto fail;
 
@@ -2401,7 +2625,7 @@ _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
         if (r < 0)
                 goto fail;
 
-        r = sd_event_add_monotonic(event, 0, 0, time_callback, bus, &bus->time_event_source);
+        r = sd_event_add_monotonic(bus->event, 0, 0, time_callback, bus, &bus->time_event_source);
         if (r < 0)
                 goto fail;
 
@@ -2409,7 +2633,7 @@ _public_ int sd_bus_attach_event(sd_bus *bus, sd_event *event, int priority) {
         if (r < 0)
                 goto fail;
 
-        r = sd_event_add_quit(event, quit_callback, bus, &bus->quit_event_source);
+        r = sd_event_add_quit(bus->event, quit_callback, bus, &bus->quit_event_source);
         if (r < 0)
                 goto fail;
 
@@ -2424,17 +2648,25 @@ _public_ int sd_bus_detach_event(sd_bus *bus) {
         assert_return(bus, -EINVAL);
         assert_return(bus->event, -ENXIO);
 
-        if (bus->input_io_event_source)
+        if (bus->input_io_event_source) {
+                sd_event_source_set_enabled(bus->input_io_event_source, SD_EVENT_OFF);
                 bus->input_io_event_source = sd_event_source_unref(bus->input_io_event_source);
+        }
 
-        if (bus->output_io_event_source)
+        if (bus->output_io_event_source) {
+                sd_event_source_set_enabled(bus->output_io_event_source, SD_EVENT_OFF);
                 bus->output_io_event_source = sd_event_source_unref(bus->output_io_event_source);
+        }
 
-        if (bus->time_event_source)
+        if (bus->time_event_source) {
+                sd_event_source_set_enabled(bus->time_event_source, SD_EVENT_OFF);
                 bus->time_event_source = sd_event_source_unref(bus->time_event_source);
+        }
 
-        if (bus->quit_event_source)
+        if (bus->quit_event_source) {
+                sd_event_source_set_enabled(bus->quit_event_source, SD_EVENT_OFF);
                 bus->quit_event_source = sd_event_source_unref(bus->quit_event_source);
+        }
 
         if (bus->event)
                 bus->event = sd_event_unref(bus->event);
@@ -2442,8 +2674,141 @@ _public_ int sd_bus_detach_event(sd_bus *bus) {
         return 0;
 }
 
-sd_bus_message* sd_bus_get_current(sd_bus *bus) {
+_public_ sd_event* sd_bus_get_event(sd_bus *bus) {
+        assert_return(bus, NULL);
+
+        return bus->event;
+}
+
+_public_ sd_bus_message* sd_bus_get_current(sd_bus *bus) {
         assert_return(bus, NULL);
 
         return bus->current;
 }
+
+static int bus_default(int (*bus_open)(sd_bus **), sd_bus **default_bus, sd_bus **ret) {
+        sd_bus *b = NULL;
+        int r;
+
+        assert(bus_open);
+        assert(default_bus);
+
+        if (!ret)
+                return !!*default_bus;
+
+        if (*default_bus) {
+                *ret = sd_bus_ref(*default_bus);
+                return 0;
+        }
+
+        r = bus_open(&b);
+        if (r < 0)
+                return r;
+
+        b->default_bus_ptr = default_bus;
+        b->tid = gettid();
+        *default_bus = b;
+
+        *ret = b;
+        return 1;
+}
+
+_public_ int sd_bus_default_system(sd_bus **ret) {
+        static __thread sd_bus *default_system_bus = NULL;
+
+        return bus_default(sd_bus_open_system, &default_system_bus, ret);
+}
+
+_public_ int sd_bus_default_user(sd_bus **ret) {
+        static __thread sd_bus *default_user_bus = NULL;
+
+        return bus_default(sd_bus_open_user, &default_user_bus, ret);
+}
+
+_public_ int sd_bus_get_tid(sd_bus *b, pid_t *tid) {
+        assert_return(b, -EINVAL);
+        assert_return(tid, -EINVAL);
+        assert_return(!bus_pid_changed(b), -ECHILD);
+
+        if (b->tid != 0) {
+                *tid = b->tid;
+                return 0;
+        }
+
+        if (b->event)
+                return sd_event_get_tid(b->event, tid);
+
+        return -ENXIO;
+}
+
+_public_ char *sd_bus_label_escape(const char *s) {
+        char *r, *t;
+        const char *f;
+
+        assert_return(s, NULL);
+
+        /* Escapes all chars that D-Bus' object path cannot deal
+         * with. Can be reversed with bus_path_unescape(). We special
+         * case the empty string. */
+
+        if (*s == 0)
+                return strdup("_");
+
+        r = new(char, strlen(s)*3 + 1);
+        if (!r)
+                return NULL;
+
+        for (f = s, t = r; *f; f++) {
+
+                /* Escape everything that is not a-zA-Z0-9. We also
+                 * escape 0-9 if it's the first character */
+
+                if (!(*f >= 'A' && *f <= 'Z') &&
+                    !(*f >= 'a' && *f <= 'z') &&
+                    !(f > s && *f >= '0' && *f <= '9')) {
+                        *(t++) = '_';
+                        *(t++) = hexchar(*f >> 4);
+                        *(t++) = hexchar(*f);
+                } else
+                        *(t++) = *f;
+        }
+
+        *t = 0;
+
+        return r;
+}
+
+_public_ char *sd_bus_label_unescape(const char *f) {
+        char *r, *t;
+
+        assert_return(f, NULL);
+
+        /* Special case for the empty string */
+        if (streq(f, "_"))
+                return strdup("");
+
+        r = new(char, strlen(f) + 1);
+        if (!r)
+                return NULL;
+
+        for (t = r; *f; f++) {
+
+                if (*f == '_') {
+                        int a, b;
+
+                        if ((a = unhexchar(f[1])) < 0 ||
+                            (b = unhexchar(f[2])) < 0) {
+                                /* Invalid escape code, let's take it literal then */
+                                *(t++) = '_';
+                        } else {
+                                *(t++) = (char) ((a << 4) | b);
+                                f += 2;
+                        }
+                } else
+                        *(t++) = *f;
+        }
+
+        *t = 0;
+
+        return r;
+}