bus-proxy: share policy between threads

This implements a shared policy cache with read-write locks. We no longer
parse the XML policy in each thread.

This will allow us to easily implement ReloadConfig().

Makefile.am

@@ -2692,6 +2692,10 @@ libsystemd_proxy_la_SOURCES = \
 	src/bus-proxyd/synthesize.c \
 	src/bus-proxyd/synthesize.h
 
+libsystemd_proxy_la_CFLAGS = \
+	$(AM_CFLAGS) \
+	-pthread
+
 libsystemd_proxy_la_LIBADD = \
 	libsystemd-internal.la \
 	libsystemd-shared.la

src/bus-proxyd/bus-proxyd.c

@@ -61,6 +61,7 @@ static char **arg_configuration = NULL;
 
 typedef struct {
         int fd;
+        SharedPolicy *policy;
 } ClientContext;
 
 static ClientContext *client_context_free(ClientContext *c) {
@@ -75,14 +76,14 @@ static ClientContext *client_context_free(ClientContext *c) {
 
 DEFINE_TRIVIAL_CLEANUP_FUNC(ClientContext*, client_context_free);
 
-static int client_context_new(ClientContext **out, int fd) {
+static int client_context_new(ClientContext **out) {
         _cleanup_(client_context_freep) ClientContext *c = NULL;
 
         c = new0(ClientContext, 1);
         if (!c)
                 return log_oom();
 
-        c->fd = fd;
+        c->fd = -1;
 
         *out = c;
         c = NULL;
@@ -105,7 +106,7 @@ static void *run_client(void *userdata) {
                 comm[sizeof(comm) - 2] = '*';
         (void) prctl(PR_SET_NAME, comm);
 
-        r = proxy_load_policy(p, arg_configuration);
+        r = proxy_set_policy(p, c->policy, arg_configuration);
         if (r < 0)
                 goto exit;
 
@@ -120,6 +121,7 @@ exit:
 }
 
 static int loop_clients(int accept_fd) {
+        _cleanup_(shared_policy_freep) SharedPolicy *sp = NULL;
         pthread_attr_t attr;
         int r;
 
@@ -135,6 +137,10 @@ static int loop_clients(int accept_fd) {
                 goto exit_attr;
         }
 
+        r = shared_policy_new(&sp);
+        if (r < 0)
+                goto exit_attr;
+
         for (;;) {
                 ClientContext *c;
                 pthread_t tid;
@@ -149,13 +155,16 @@ static int loop_clients(int accept_fd) {
                         break;
                 }
 
-                r = client_context_new(&c, fd);
+                r = client_context_new(&c);
                 if (r < 0) {
                         log_oom();
                         close(fd);
                         continue;
                 }
 
+                c->fd = fd;
+                c->policy = sp;
+
                 r = pthread_create(&tid, &attr, run_client, c);
                 if (r < 0) {
                         log_error("Cannot spawn thread: %m");

src/bus-proxyd/bus-xml-policy.c

@@ -1106,6 +1106,122 @@ void policy_dump(Policy *p) {
         fflush(stdout);
 }
 
+int shared_policy_new(SharedPolicy **out) {
+        SharedPolicy *sp;
+        int r;
+
+        sp = new0(SharedPolicy, 1);
+        if (!sp)
+                return log_oom();
+
+        r = pthread_mutex_init(&sp->lock, NULL);
+        if (r < 0) {
+                log_error_errno(r, "Cannot initialize shared policy mutex: %m");
+                goto exit_free;
+        }
+
+        r = pthread_rwlock_init(&sp->rwlock, NULL);
+        if (r < 0) {
+                log_error_errno(r, "Cannot initialize shared policy rwlock: %m");
+                goto exit_mutex;
+        }
+
+        *out = sp;
+        sp = NULL;
+        return 0;
+
+        /* pthread lock destruction is not fail-safe... meh! */
+exit_mutex:
+        pthread_mutex_destroy(&sp->lock);
+exit_free:
+        free(sp);
+        return r;
+}
+
+SharedPolicy *shared_policy_free(SharedPolicy *sp) {
+        if (!sp)
+                return NULL;
+
+        policy_free(sp->policy);
+        pthread_rwlock_destroy(&sp->rwlock);
+        pthread_mutex_destroy(&sp->lock);
+        free(sp);
+
+        return NULL;
+}
+
+static int shared_policy_reload_unlocked(SharedPolicy *sp, char **configuration) {
+        Policy old, buffer = {};
+        bool free_old;
+        int r;
+
+        assert(sp);
+
+        r = policy_load(&buffer, configuration);
+        if (r < 0)
+                return log_error_errno(r, "Failed to load policy: %m");
+
+        /* policy_dump(&buffer); */
+
+        pthread_rwlock_wrlock(&sp->rwlock);
+        memcpy(&old, &sp->buffer, sizeof(old));
+        memcpy(&sp->buffer, &buffer, sizeof(buffer));
+        free_old = !!sp->policy;
+        sp->policy = &sp->buffer;
+        pthread_rwlock_unlock(&sp->rwlock);
+
+        if (free_old)
+                policy_free(&old);
+
+        return 0;
+}
+
+int shared_policy_reload(SharedPolicy *sp, char **configuration) {
+        int r;
+
+        assert(sp);
+
+        pthread_mutex_lock(&sp->lock);
+        r = shared_policy_reload_unlocked(sp, configuration);
+        pthread_mutex_unlock(&sp->lock);
+
+        return r;
+}
+
+int shared_policy_preload(SharedPolicy *sp, char **configuration) {
+        int r;
+
+        assert(sp);
+
+        pthread_mutex_lock(&sp->lock);
+        if (!sp->policy)
+                r = shared_policy_reload_unlocked(sp, configuration);
+        else
+                r = 0;
+        pthread_mutex_unlock(&sp->lock);
+
+        return r;
+}
+
+Policy *shared_policy_acquire(SharedPolicy *sp) {
+        assert(sp);
+
+        pthread_rwlock_rdlock(&sp->rwlock);
+        if (sp->policy)
+                return sp->policy;
+        pthread_rwlock_unlock(&sp->rwlock);
+
+        return NULL;
+}
+
+void shared_policy_release(SharedPolicy *sp, Policy *p) {
+        assert(sp);
+        assert(!p || sp->policy == p);
+
+        if (p)
+                pthread_rwlock_unlock(&sp->rwlock);
+}
+
 static const char* const policy_item_type_table[_POLICY_ITEM_TYPE_MAX] = {
         [_POLICY_ITEM_TYPE_UNSET] = "unset",
         [POLICY_ITEM_ALLOW] = "allow",

src/bus-proxyd/bus-xml-policy.h

@@ -22,6 +22,7 @@
 ***/
 
 #include <inttypes.h>
+#include <pthread.h>
 
 #include "list.h"
 #include "hashmap.h"
@@ -75,6 +76,15 @@ typedef struct Policy {
         Hashmap *group_items;
 } Policy;
 
+typedef struct SharedPolicy {
+        pthread_mutex_t lock;
+        pthread_rwlock_t rwlock;
+        Policy buffer;
+        Policy *policy;
+} SharedPolicy;
+
+/* policy */
+
 int policy_load(Policy *p, char **files);
 void policy_free(Policy *p);
 
@@ -106,3 +116,15 @@ PolicyItemType policy_item_type_from_string(const char *s) _pure_;
 
 const char* policy_item_class_to_string(PolicyItemClass t) _const_;
 PolicyItemClass policy_item_class_from_string(const char *s) _pure_;
+
+/* shared policy */
+
+int shared_policy_new(SharedPolicy **out);
+SharedPolicy *shared_policy_free(SharedPolicy *sp);
+
+int shared_policy_reload(SharedPolicy *sp, char **configuration);
+int shared_policy_preload(SharedPolicy *sp, char **configuration);
+Policy *shared_policy_acquire(SharedPolicy *sp);
+void shared_policy_release(SharedPolicy *sp, Policy *p);
+
+DEFINE_TRIVIAL_CLEANUP_FUNC(SharedPolicy*, shared_policy_free);

src/bus-proxyd/driver.c

@@ -80,7 +80,7 @@ static int get_creds_by_message(sd_bus *bus, sd_bus_message *m, uint64_t mask, s
         return get_creds_by_name(bus, name, mask, _creds, error);
 }
 
-int bus_proxy_process_driver(sd_bus *a, sd_bus *b, sd_bus_message *m, Policy *policy, const struct ucred *ucred, Set *owned_names) {
+int bus_proxy_process_driver(sd_bus *a, sd_bus *b, sd_bus_message *m, SharedPolicy *sp, const struct ucred *ucred, Set *owned_names) {
         int r;
 
         assert(a);
@@ -455,8 +455,16 @@ int bus_proxy_process_driver(sd_bus *a, sd_bus *b, sd_bus_message *m, Policy *po
                 if (r < 0)
                         return synthetic_reply_method_errno(m, r, NULL);
 
-                if (policy && !policy_check_own(policy, ucred->uid, ucred->gid, name))
-                        return synthetic_reply_method_errno(m, -EPERM, NULL);
+                if (sp) {
+                        Policy *policy;
+                        bool denied;
+
+                        policy = shared_policy_acquire(sp);
+                        denied = !policy_check_own(policy, ucred->uid, ucred->gid, name);
+                        shared_policy_release(sp, policy);
+                        if (denied)
+                                return synthetic_reply_method_errno(m, -EPERM, NULL);
+                }
 
                 if ((flags & ~(BUS_NAME_ALLOW_REPLACEMENT|BUS_NAME_REPLACE_EXISTING|BUS_NAME_DO_NOT_QUEUE)) != 0)
                         return synthetic_reply_method_errno(m, -EINVAL, NULL);

src/bus-proxyd/driver.h

@@ -24,4 +24,4 @@
 #include "sd-bus.h"
 #include "bus-xml-policy.h"
 
-int bus_proxy_process_driver(sd_bus *a, sd_bus *b, sd_bus_message *m, Policy *policy, const struct ucred *ucred, Set *owned_names);
+int bus_proxy_process_driver(sd_bus *a, sd_bus *b, sd_bus_message *m, SharedPolicy *sp, const struct ucred *ucred, Set *owned_names);

src/bus-proxyd/proxy.c

@@ -255,16 +255,27 @@ Proxy *proxy_free(Proxy *p) {
         return NULL;
 }
 
-int proxy_load_policy(Proxy *p, char **configuration) {
+int proxy_set_policy(Proxy *p, SharedPolicy *sp, char **configuration) {
         _cleanup_strv_free_ char **strv = NULL;
+        Policy *policy;
         int r;
 
         assert(p);
+        assert(sp);
 
         /* no need to load legacy policy if destination is not kdbus */
         if (!p->dest_bus->is_kernel)
                 return 0;
 
+        p->policy = sp;
+
+        policy = shared_policy_acquire(sp);
+        if (policy) {
+                /* policy already pre-loaded */
+                shared_policy_release(sp, policy);
+                return 0;
+        }
+
         if (!configuration) {
                 const char *scope;
 
@@ -291,30 +302,30 @@ int proxy_load_policy(Proxy *p, char **configuration) {
                 configuration = strv;
         }
 
-        r = policy_load(&p->policy_buffer, configuration);
-        if (r < 0)
-                return log_error_errno(r, "Failed to load policy: %m");
-
-        p->policy = &p->policy_buffer;
-        /* policy_dump(p->policy); */
-
-        return 0;
+        return shared_policy_preload(sp, configuration);
 }
 
 int proxy_hello_policy(Proxy *p, uid_t original_uid) {
+        Policy *policy;
+        int r = 0;
+
         assert(p);
 
         if (!p->policy)
                 return 0;
 
+        policy = shared_policy_acquire(p->policy);
+
         if (p->local_creds.uid == original_uid)
                 log_debug("Permitting access, since bus owner matches bus client.");
-        else if (policy_check_hello(p->policy, p->local_creds.uid, p->local_creds.gid))
+        else if (policy_check_hello(policy, p->local_creds.uid, p->local_creds.gid))
                 log_debug("Permitting access due to XML policy.");
         else
-                return log_error_errno(EPERM, "Policy denied connection.");
+                r = log_error_errno(EPERM, "Policy denied connection.");
 
-        return 0;
+        shared_policy_release(p->policy, policy);
+
+        return r;
 }
 
 static int proxy_wait(Proxy *p) {
@@ -384,7 +395,7 @@ static int handle_policy_error(sd_bus_message *m, int r) {
         return r;
 }
 
-static int process_policy(sd_bus *from, sd_bus *to, sd_bus_message *m, Policy *policy, const struct ucred *our_ucred, Set *owned_names) {
+static int process_policy_unlocked(sd_bus *from, sd_bus *to, sd_bus_message *m, Policy *policy, const struct ucred *our_ucred, Set *owned_names) {
         int r;
 
         assert(from);
@@ -584,6 +595,19 @@ static int process_policy(sd_bus *from, sd_bus *to, sd_bus_message *m, Policy *p
         return 0;
 }
 
+static int process_policy(sd_bus *from, sd_bus *to, sd_bus_message *m, SharedPolicy *sp, const struct ucred *our_ucred, Set *owned_names) {
+        Policy *policy;
+        int r;
+
+        assert(sp);
+
+        policy = shared_policy_acquire(sp);
+        r = process_policy_unlocked(from, to, m, policy, our_ucred, owned_names);
+        shared_policy_release(sp, policy);
+
+        return r;
+}
+
 static int process_hello(Proxy *p, sd_bus_message *m) {
         _cleanup_bus_message_unref_ sd_bus_message *n = NULL;
         bool is_hello;

src/bus-proxyd/proxy.h

@@ -37,8 +37,7 @@ struct Proxy {
         sd_bus *dest_bus;
 
         Set *owned_names;
-        Policy policy_buffer;
-        Policy *policy;
+        SharedPolicy *policy;
 
         bool got_hello : 1;
 };
@@ -46,7 +45,7 @@ struct Proxy {
 int proxy_new(Proxy **out, int in_fd, int out_fd, const char *dest);
 Proxy *proxy_free(Proxy *p);
 
-int proxy_load_policy(Proxy *p, char **configuration);
+int proxy_set_policy(Proxy *p, SharedPolicy *policy, char **configuration);
 int proxy_hello_policy(Proxy *p, uid_t original_uid);
 int proxy_run(Proxy *p);
 

src/bus-proxyd/stdio-bridge.c

@@ -54,7 +54,6 @@
 
 static char *arg_address = NULL;
 static char *arg_command_line_buffer = NULL;
-static char **arg_configuration = NULL;
 
 static int help(void) {
 
@@ -62,7 +61,6 @@ static int help(void) {
                "Connect STDIO to a given bus address.\n\n"
                "  -h --help               Show this help\n"
                "     --version            Show package version\n"
-               "     --configuration=PATH Configuration file or directory\n"
                "     --machine=MACHINE    Connect to specified machine\n"
                "     --address=ADDRESS    Connect to the bus specified by ADDRESS\n"
                "                          (default: " DEFAULT_SYSTEM_BUS_ADDRESS ")\n",
@@ -76,7 +74,6 @@ static int parse_argv(int argc, char *argv[]) {
         enum {
                 ARG_VERSION = 0x100,
                 ARG_ADDRESS,
-                ARG_CONFIGURATION,
                 ARG_MACHINE,
         };
 
@@ -84,12 +81,11 @@ static int parse_argv(int argc, char *argv[]) {
                 { "help",            no_argument,       NULL, 'h'                 },
                 { "version",         no_argument,       NULL, ARG_VERSION         },
                 { "address",         required_argument, NULL, ARG_ADDRESS         },
-                { "configuration",   required_argument, NULL, ARG_CONFIGURATION   },
                 { "machine",         required_argument, NULL, ARG_MACHINE         },
                 {},
         };
 
-        int c, r;
+        int c;
 
         assert(argc >= 0);
         assert(argv);
@@ -119,12 +115,6 @@ static int parse_argv(int argc, char *argv[]) {
                         break;
                 }
 
-                case ARG_CONFIGURATION:
-                        r = strv_extend(&arg_configuration, optarg);
-                        if (r < 0)
-                                return log_oom();
-                        break;
-
                 case ARG_MACHINE: {
                         _cleanup_free_ char *e = NULL;
                         char *a;
@@ -256,14 +246,6 @@ int main(int argc, char *argv[]) {
         if (r < 0)
                 goto finish;
 
-        r = proxy_load_policy(p, arg_configuration);
-        if (r < 0)
-                goto finish;
-
-        r = proxy_hello_policy(p, getuid());
-        if (r < 0)
-                goto finish;
-
         r = rename_service(p->dest_bus, p->local_bus);
         if (r < 0)
                 log_debug_errno(r, "Failed to rename process: %m");
@@ -275,7 +257,6 @@ finish:
                   "STOPPING=1\n"
                   "STATUS=Shutting down.");
 
-        strv_free(arg_configuration);
         free(arg_address);
 
         return r < 0 ? EXIT_FAILURE : EXIT_SUCCESS;