diff --git a/doc/build.md b/doc/build.md
index 298437b6..5deb0ec8 100644
--- a/doc/build.md
+++ b/doc/build.md
@@ -44,7 +44,8 @@ Below are a few of the main switches to configure:
 
 * `--enable-static`: Build Finit statically.  The plugins will be
   built-ins (.o files) and all external libraries, except the C library
-  will be linked statically.
+  will be linked statically.  Privileged D-Bus methods then accept only
+  `root`, see [Authorization](dbus.md#authorization)
 
 * `--enable-kernel-cmdline`: Enable Finit pre-4.1 parsing of init args from
   `/proc/cmdline`, this is *not recommended* since Finit may be running as the
diff --git a/doc/dbus.md b/doc/dbus.md
index e7453118..15118946 100644
--- a/doc/dbus.md
+++ b/doc/dbus.md
@@ -204,20 +204,37 @@ conditions belong to Finit's state machine.
 Authorization
 -------------
 
-Privileged methods reject any caller whose peer `uid` isn't 0.  On the
-**local** bus the kernel's `SO_PEERCRED` socket option tells Finit exactly
-who's calling, so privilege escalation through the bus is impossible.
+Privileged methods accept `root`, and any caller belonging to the group given
+to `--with-group` at build time.  That is the same set the socket mode already
+admits, so the two gates agree instead of the socket letting the group in and
+every method turning it away.
 
-On the **system** bus, all incoming traffic is treated as unprivileged: it
-arrives through `dbus-daemon` (typically running as root) and Finit cannot yet
-ask the daemon for the real requester's uid via `GetConnectionUnixUser`.  This
-means external tooling can freely `Get`/`Introspect`/`ListServices`, but every
-state-changing method returns `org.freedesktop.DBus.Error.AccessDenied`.
-Per-sender uid lookup is on the roadmap.
+A build configured with `--enable-static` accepts only `root`.  Group
+membership comes from NSS, which the C library loads with `dlopen()`, and a
+build meant to link statically cannot count on that, so the lookup is compiled
+out rather than left to fail open.  The socket mode is unchanged, so the group
+can still connect, it just cannot invoke a privileged method.
+
+On the **local** bus the kernel decides this at `connect()`, supplementary
+groups included, and `SO_PEERCRED` tells Finit exactly who is calling, so
+privilege escalation through the bus is impossible.
+
+On the **system** bus one connection carries every caller, so `SO_PEERCRED`
+describes `dbus-daemon` rather than whoever asked.  For a privileged method
+Finit asks the bus driver `GetConnectionUnixUser` about the message sender and
+holds the call until the answer arrives.  Nothing blocks: the reply comes back
+through the same event loop as everything else, and the held call is then
+dispatched or refused on its merits.
+
+Answers are cached per sender.  A bus never reuses a unique name while it
+runs, so an answer holds for as long as that bus does; Finit empties the cache
+when the broker goes away, since a new one numbers its clients from scratch.
+A caller Finit cannot identify is refused, so the failure mode is a denial
+rather than an escalation.
 
 When a privileged method is rejected the error name is exactly
 `org.freedesktop.DBus.Error.AccessDenied`, and the body carries a short reason
-string (e.g. `"permission denied: Start requires root"`).
+string.
 
 `initctl` integration
 ---------------------
diff --git a/libink/client.c b/libink/client.c
index 15a1760d..6a0c60b4 100644
--- a/libink/client.c
+++ b/libink/client.c
@@ -171,15 +171,8 @@ static int read_one(link_client_t *c, struct link_msg *msg)
 
 static void publish_reply(link_client_t *c, const struct link_msg *m)
 {
-	c->reply.type       = m->type;
-	c->reply.signature  = m->signature;
-	c->reply.error_name = m->error_name;
-	c->reply.path       = m->path;
-	c->reply.interface  = m->interface;
-	c->reply.member     = m->member;
-	c->reply.body       = m->body_avail ? m->body : NULL;
-	c->reply.body_len   = m->body_avail;
-	c->have_reply       = 1;
+	__msg_to_reply(&c->reply, m);
+	c->have_reply = 1;
 }
 
 /* The reply view in c->reply points into c->rxbuf and is invalidated
@@ -261,12 +254,7 @@ int link_client_call(link_client_t *c,
 		     const char *signature,
 		     const uint8_t *body, size_t body_len)
 {
-	/* Generous: Manager1 headers fit in ~150 B, but the buffer is
-	 * shared with whatever future callers throw at us, and an
-	 * overflow only manifests as a silent LINK_CALL_FAIL via
-	 * __msg_build_method_call returning -1.  1 KiB on stack
-	 * is cheap insurance. */
-	uint8_t  hdr[1024];
+	uint8_t  hdr[LINK_CALL_HDR_MAX];
 	ssize_t  hlen;
 	uint32_t serial;
 
@@ -328,54 +316,20 @@ int link_reply_get_u32(const link_reply_t *r, uint32_t *out)
 	return link_r_u32(&reader, out);
 }
 
-/* Marshal varargs into `body` (capacity `cap`) according to `sig`.
- * Returns the marshalled length on success, -1 on overflow or
- * unsupported type code. */
-static ssize_t marshal_va(uint8_t *body, size_t cap,
-			  const char *sig, va_list ap)
-{
-	link_writer_t w;
-	const char   *s;
-
-	link_writer_init(&w, body, cap);
-	for (s = sig; *s; s++) {
-		switch (*s) {
-		case 'y':
-			link_w_byte(&w, (uint8_t)va_arg(ap, int));
-			break;
-		case 'b':
-			link_w_bool(&w, va_arg(ap, int));
-			break;
-		case 'u':
-			link_w_u32(&w, va_arg(ap, uint32_t));
-			break;
-		case 's':
-			link_w_string(&w, va_arg(ap, const char *));
-			break;
-		case 'o':
-			link_w_path(&w, va_arg(ap, const char *));
-			break;
-		default:
-			return -1;
-		}
-	}
-	return link_writer_finish(&w);
-}
-
 int link_client_call_v(link_client_t *c,
 		       const char *obj_path,
 		       const char *interface,
 		       const char *member,
 		       const char *signature, ...)
 {
-	uint8_t body[1024];
+	uint8_t body[LINK_CALL_BODY_MAX];
 	ssize_t body_len = 0;
 
 	if (signature && *signature) {
 		va_list ap;
 
 		va_start(ap, signature);
-		body_len = marshal_va(body, sizeof(body), signature, ap);
+		body_len = __marshal_va(body, sizeof(body), signature, ap);
 		va_end(ap);
 		if (body_len < 0)
 			return LINK_CALL_FAIL;
diff --git a/libink/connection.c b/libink/connection.c
index f88afda8..3fe799bc 100644
--- a/libink/connection.c
+++ b/libink/connection.c
@@ -5,6 +5,7 @@
  */
 
 #include <errno.h>
+#include <stdarg.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
@@ -21,6 +22,70 @@ uid_t link_connection_get_uid(const link_connection_t *conn)
 	return conn ? conn->peer_uid : (uid_t)-1;
 }
 
+/* Issue a method call and remember the serial so the reply can be
+ * handed back to `cb` when the read loop picks it up.  Nothing here
+ * waits: this is the counterpart of link_client_call() for a
+ * connection already owned by the event loop. */
+int link_connection_call(link_connection_t *conn, const char *destination,
+			 const char *path, const char *interface, const char *member,
+			 link_reply_cb_t cb, void *userdata,
+			 const char *signature, ...)
+{
+	uint8_t  body[LINK_CALL_BODY_MAX];
+	uint8_t  hdr[LINK_CALL_HDR_MAX];
+	ssize_t  blen = 0;
+	ssize_t  hlen;
+	uint32_t serial;
+	int      i;
+
+	if (!conn || conn->fd < 0 || !path || !member) {
+		errno = EINVAL;
+		return -1;
+	}
+
+	for (i = 0; i < LINK_PENDING_CAP; i++) {
+		if (!conn->pending[i].used)
+			break;
+	}
+	if (i == LINK_PENDING_CAP) {
+		errno = EBUSY;
+		return -1;
+	}
+
+	if (signature && *signature) {
+		va_list ap;
+
+		va_start(ap, signature);
+		blen = __marshal_va(body, sizeof(body), signature, ap);
+		va_end(ap);
+		if (blen < 0) {
+			errno = EMSGSIZE;
+			return -1;
+		}
+	}
+
+	serial = ++conn->next_serial;
+	hlen = __msg_build_method_call(hdr, sizeof(hdr), serial, path, interface,
+				       member, destination, signature, (uint32_t)blen);
+	if (hlen < 0) {
+		errno = EMSGSIZE;
+		return -1;
+	}
+
+	if (__io_write_all(conn->fd, hdr, (size_t)hlen) < 0)
+		return -1;
+	if (blen > 0 && __io_write_all(conn->fd, body, (size_t)blen) < 0)
+		return -1;
+
+
+	conn->pending[i].used     = 1;
+	conn->pending[i].serial   = serial;
+	conn->pending[i].cb       = cb;
+	conn->pending[i].userdata = userdata;
+
+	return 0;
+}
+
 void link_connection_close(link_connection_t *conn)
 {
 	size_t i;
@@ -28,6 +93,11 @@ void link_connection_close(link_connection_t *conn)
 	if (!conn)
 		return;
 
+
+	/* Anything waiting on this connection has to be told, or a parked
+	 * call sits forever and its caller never hears back. */
+	__dispatch_forget_conn(conn);
+
 	for (i = 0; i < conn->matches_count; i++)
 		__match_free(conn->matches[i]);
 
@@ -52,7 +122,7 @@ static int process_binary(link_connection_t *conn)
 		if (consumed < 0)
 			return -1;
 
-		if (__dispatch_message(conn, &msg) < 0)
+		if (__dispatch_message(conn, &msg, (size_t)consumed) < 0)
 			return -1;
 
 		memmove(conn->rxbuf, conn->rxbuf + consumed,
diff --git a/libink/dispatch.c b/libink/dispatch.c
index 1bfb59ef..44642d7b 100644
--- a/libink/dispatch.c
+++ b/libink/dispatch.c
@@ -260,7 +260,7 @@ int __send_error(link_connection_t *conn, const struct link_msg *req,
 const char *link_call_path     (const link_call_t *c) { return c ? c->incoming.path      : NULL; }
 const char *link_call_interface(const link_call_t *c) { return c ? c->incoming.interface : NULL; }
 const char *link_call_member   (const link_call_t *c) { return c ? c->incoming.member    : NULL; }
-uid_t       link_call_uid      (const link_call_t *c) { return c ? c->conn->peer_uid     : (uid_t)-1; }
+uid_t       link_call_uid      (const link_call_t *c) { return c ? c->uid               : LINK_UID_UNKNOWN; }
 
 link_writer_t *link_call_reply(link_call_t *call)
 {
@@ -324,20 +324,213 @@ size_t link_r_pos (const link_reader_t *r)            { return r->off;
 
 /* ----------  dispatch entry point  ---------- */
 
-int __dispatch_message(link_connection_t *conn, const struct link_msg *m)
+/* ----------  replies to our own outbound calls  ---------- */
+
+/* Hand a reply to whoever issued the matching link_connection_call().
+ * Unmatched replies are dropped: a broker is free to send us things we
+ * never asked for, and that is not a reason to drop the connection. */
+static void deliver_reply(link_connection_t *conn, const struct link_msg *m)
+{
+	link_reply_cb_t  cb;
+	link_reply_t     r;
+	void            *userdata;
+	int              i;
+
+	for (i = 0; i < LINK_PENDING_CAP; i++) {
+		if (conn->pending[i].used && conn->pending[i].serial == m->reply_serial)
+			break;
+	}
+	if (i == LINK_PENDING_CAP) {
+		return;
+	}
+
+	cb       = conn->pending[i].cb;
+	userdata = conn->pending[i].userdata;
+	conn->pending[i].used = 0;
+
+	if (!cb)
+		return;
+
+	__msg_to_reply(&r, m);
+	cb(conn, &r, userdata);
+}
+
+/* ----------  calls parked while their caller is identified  ---------- */
+
+/* Every park gets a token that is never issued twice, so a resolver
+ * answering late, twice, or after its connection went away resumes
+ * nothing rather than whatever call has since taken the slot. */
+static struct link_parked *park(link_connection_t *conn, const uint8_t *frame,
+				size_t len, link_authz_t *tok)
+{
+	link_server_t *srv = conn->server;
+	int i;
+
+	if (!frame || !len || len > LINK_PARKED_MSG_MAX)
+		return NULL;
+
+	for (i = 0; i < LINK_PARKED_CAP; i++) {
+		if (!srv->parked[i].tok)
+			break;
+	}
+	if (i == LINK_PARKED_CAP)
+		return NULL;
+
+	srv->parked[i].tok  = ++srv->next_tok;
+	srv->parked[i].conn = conn;
+	srv->parked[i].len  = len;
+	memcpy(srv->parked[i].buf, frame, len);
+	*tok = srv->parked[i].tok;
+
+	return &srv->parked[i];
+}
+
+static void unpark(struct link_parked *p)
+{
+	p->tok  = 0;
+	p->conn = NULL;
+}
+
+void __dispatch_forget_conn(link_connection_t *conn)
+{
+	link_server_t *srv = conn->server;
+	int i;
+
+	/* Drop parked calls first.  A pending callback below may try to
+	 * resolve one, and resuming a dispatch on a connection that is
+	 * being torn down is no use to anyone; an invalidated slot makes
+	 * that resolve a no-op instead. */
+	if (srv) {
+		for (i = 0; i < LINK_PARKED_CAP; i++) {
+			if (srv->parked[i].conn == conn)
+				unpark(&srv->parked[i]);
+		}
+	}
+
+	for (i = 0; i < LINK_PENDING_CAP; i++) {
+		if (conn->pending[i].used && conn->pending[i].cb)
+			conn->pending[i].cb(conn, NULL, conn->pending[i].userdata);
+		conn->pending[i].used = 0;
+	}
+}
+
+static int dispatch_call(link_connection_t *conn, const struct link_msg *m,
+			 const uint8_t *frame, size_t framelen,
+			 const uid_t *known_uid);
+
+void link_uid_resolved(link_server_t *server, link_authz_t tok, uid_t uid)
+{
+	uint8_t             buf[LINK_PARKED_MSG_MAX];
+	struct link_parked *p = NULL;
+	link_connection_t  *conn;
+	struct link_msg     msg;
+	size_t              len;
+	int                 i;
+
+	if (!server || !tok)
+		return;
+
+	for (i = 0; i < LINK_PARKED_CAP; i++) {
+		if (server->parked[i].tok == tok) {
+			p = &server->parked[i];
+			break;
+		}
+	}
+	if (!p)
+		return;		/* stale handle, already answered */
+
+	/* Copy the message out and free the slot before dispatching:
+	 * the handler may park a call of its own. */
+	conn = p->conn;
+	len  = p->len;
+	memcpy(buf, p->buf, len);
+	unpark(p);
+
+	if (!conn || __msg_parse(buf, len, &msg) <= 0)
+		return;
+
+	(void)dispatch_call(conn, &msg, NULL, 0, &uid);
+}
+
+int __dispatch_message(link_connection_t *conn, const struct link_msg *m, size_t framelen)
+{
+	if (m->type == LINK_MSG_METHOD_RETURN || m->type == LINK_MSG_ERROR) {
+		deliver_reply(conn, m);
+		return 0;
+	}
+
+	if (m->type != LINK_MSG_METHOD_CALL) {
+		/* Signals from a client to PID 1 are nonsense; drop. */
+		return 0;
+	}
+
+	return dispatch_call(conn, m, conn->rxbuf, framelen, NULL);
+}
+
+/* Who may invoke a privileged method.  Without an authorizer, only
+ * root, which is what libink can decide on its own. */
+static int caller_may(link_server_t *srv, uid_t uid)
+{
+	if (uid == LINK_UID_UNKNOWN)
+		return 0;
+	if (srv && srv->authorizer)
+		return srv->authorizer(uid, srv->authz_userdata);
+
+	return uid == 0;
+}
+
+/* Ask who is calling on a broker connection, where the message is the
+ * only evidence.  Returns 0 with *uid set, 1 when the call was parked
+ * and will be dispatched again once the resolver answers, -1 when the
+ * caller cannot be identified, and -2 when we have no room to ask. */
+#define CALLER_UID_BUSY (-2)
+
+static int resolve_caller(link_connection_t *conn, const struct link_msg *m,
+			  const uint8_t *frame, size_t framelen, uid_t *uid)
+{
+	link_server_t      *srv = conn->server;
+	struct link_parked *p;
+	link_authz_t        tok;
+	int                 rc;
+
+	if (!srv || !srv->uid_resolver || !m->sender)
+		return -1;
+
+	/* Enforce the rule link.h states, rather than trusting every
+	 * resolver to remember it: a truncated sender key would let two
+	 * callers share one identity. */
+	if (strlen(m->sender) >= LINK_SENDER_MAX) {
+		return -1;
+	}
+
+	/* Park first so the resolver has somewhere to answer, then let
+	 * it release the slot immediately if it already knew. */
+	p = park(conn, frame, framelen, &tok);
+	if (!p)
+		return CALLER_UID_BUSY;
+
+	rc = srv->uid_resolver(conn, m->sender, tok, uid, srv->uid_userdata);
+	if (rc != 1)
+		unpark(p);
+
+	return rc;
+}
+
+static int dispatch_call(link_connection_t *conn, const struct link_msg *m,
+			 const uint8_t *frame, size_t framelen,
+			 const uid_t *known_uid)
 {
 	struct link_object       *o;
 	struct link_vtable_entry *e = NULL;
 	const link_method_t      *meth;
 	struct link_call          call;
 	ssize_t                  blen;
+	uid_t                    call_uid;
 	int                      rc;
 
-	if (m->type != LINK_MSG_METHOD_CALL) {
-		/* Signals and replies from a client to PID 1 are nonsense;
-		 * silently drop. */
-		return 0;
-	}
+	/* What a handler sees via link_call_uid().  Unresolved on a broker
+	 * connection until a privileged method forces the question. */
+	call_uid = known_uid ? *known_uid : conn->peer_uid;
 
 	if (!m->path || !m->member) {
 		return __send_error(conn, m,
@@ -345,6 +538,7 @@ int __dispatch_message(link_connection_t *conn, const struct link_msg *m)
 			"Method call without path or member");
 	}
 
+
 	/* Built-in DBus interfaces (Hello, Ping, Introspect, Properties)
 	 * are handled here before object-tree lookup, which means they
 	 * also run before the LINK_METHOD_PRIVILEGED authz gate further
@@ -374,24 +568,47 @@ int __dispatch_message(link_connection_t *conn, const struct link_msg *m)
 		const char *got = m->signature ? m->signature : "";
 		const char *want = meth->in_sig ? meth->in_sig : "";
 
-		if (strcmp(got, want) != 0)
+		if (strcmp(got, want) != 0) {
 			return __send_error(conn, m,
 				"org.freedesktop.DBus.Error.InvalidArgs",
 				"Argument signature mismatch");
+		}
 	}
 
-	/* Per-method authorization.  PRIVILEGED methods require uid 0;
-	 * the peer's uid was captured via SO_PEERCRED at accept time
-	 * and verified against the AUTH EXTERNAL claim, so we can trust
-	 * conn->peer_uid here. */
-	if ((meth->flags & LINK_METHOD_PRIVILEGED) && conn->peer_uid != 0) {
-		return __send_error(conn, m,
-			"org.freedesktop.DBus.Error.AccessDenied",
-			"Method requires root privileges");
+	/* Per-method authorization.  PRIVILEGED methods require uid 0.
+	 * On an ordinary connection the peer's uid was captured via
+	 * SO_PEERCRED at accept time and verified against the AUTH
+	 * EXTERNAL claim, so conn->peer_uid is the answer.  A broker
+	 * connection carries every caller at once, so who is asking has
+	 * to be established per message, which may park the call. */
+	if (meth->flags & LINK_METHOD_PRIVILEGED) {
+		if (conn->broker && !known_uid) {
+			rc = resolve_caller(conn, m, frame, framelen, &call_uid);
+			if (rc == 1)
+				return 0;	/* parked, resumed later */
+
+			if (rc == CALLER_UID_BUSY) {
+				/* Not a permission problem: root may well
+				 * be asking, we just have no slot to find
+				 * out in.  Say so, it is retryable. */
+				return __send_error(conn, m,
+					"org.freedesktop.DBus.Error.LimitsExceeded",
+					"Too many calls awaiting authorization");
+			}
+			if (rc < 0)
+				call_uid = (uid_t)-1;
+		}
+
+		if (!caller_may(conn->server, call_uid)) {
+			return __send_error(conn, m,
+				"org.freedesktop.DBus.Error.AccessDenied",
+				"Caller is not privileged for this method");
+		}
 	}
 
 	memset(&call, 0, sizeof(call));
 	call.conn     = conn;
+	call.uid      = call_uid;
 	call.incoming = *m;
 	__r_init(&call.read_cursor, m->body, m->body_avail);
 
diff --git a/libink/internal.h b/libink/internal.h
index b48c3e51..5f0f778a 100644
--- a/libink/internal.h
+++ b/libink/internal.h
@@ -24,9 +24,22 @@ typedef enum {
 #define LINK_AUTH_LINEBUF_SIZE 256
 #define LINK_RX_BUF_SIZE       (64 * 1024)
 #define LINK_TX_BUF_SIZE       (16 * 1024)
-#define LINK_UNIQUE_NAME_LEN   16
+#define LINK_UNIQUE_NAME_LEN   LINK_SENDER_MAX
 #define LINK_MATCH_RULE_MAX    256	/* per-peer match rule cap */
 #define LINK_MATCH_PEER_CAP    16	/* max active match rules per peer */
+#define LINK_PENDING_CAP        4	/* outbound calls awaiting a reply */
+/* Staging for an outgoing method call.  Generous on purpose: headers
+ * for the calls libink makes run to ~150 B, and both the synchronous
+ * and the connection-side path build into these, so one answer rather
+ * than a number per call site. */
+#define LINK_CALL_HDR_MAX    1024
+#define LINK_CALL_BODY_MAX   1024
+#define LINK_PARKED_CAP         4	/* inbound calls awaiting a uid */
+/* A call parked for authorization is a privileged one: an object path
+ * and at most a service name.  Finit's per-service paths alone run to
+ * 512 bytes, so leave room for the header around one.  Anything that
+ * does not fit is denied rather than held. */
+#define LINK_PARKED_MSG_MAX  1024
 
 /* Per-vtable record attached to an object's interface list. */
 struct link_vtable_entry {
@@ -46,19 +59,44 @@ struct link_object {
 
 TAILQ_HEAD(link_object_list, link_object);
 
+/* An inbound method call held while we find out who sent it.  The
+ * message is copied because rxbuf is reused as soon as we return to
+ * the read loop.  `tok` is the handle the resolver answers with, and
+ * zero when the slot is free. */
+struct link_parked {
+	link_authz_t        tok;
+	link_connection_t  *conn;
+	size_t              len;
+	uint8_t             buf[LINK_PARKED_MSG_MAX];
+};
+
 struct link_server {
 	int                       fd;
 	char                      path[LINK_PATH_MAX];
 	struct link_object_list    objects;
 	uint32_t                  next_unique_id;	/* for ":1.N" names */
+
+	/* Set by link_server_set_uid_resolver(); see link.h. */
+	link_uid_resolver_t       uid_resolver;
+	void                     *uid_userdata;
+
+	/* Set by link_server_set_authorizer(); see link.h. */
+	link_authorizer_t         authorizer;
+	void                     *authz_userdata;
+	struct link_parked        parked[LINK_PARKED_CAP];
+	link_authz_t              next_tok;
 };
 
 /* The reply being assembled inside a method handler.
  *
  * The reply body lives in conn->txbuf, not on this struct, so a
  * stack-allocated link_call (in dispatch) stays small.  Sharing the
- * connection's txbuf is safe: the event loop is single-threaded and
- * a connection only ever has one in-flight method call at a time. */
+ * connection's txbuf is safe because a reply is marshalled and sent
+ * without yielding.  Note that parking means several calls can be in
+ * flight on one connection: what is held is the request, and
+ * link_uid_resolved() resumes from a copy, so txbuf is still only
+ * ever used by one reply at a time.  An async handler that returned
+ * before writing its reply would break that. */
 struct link_call {
 	link_connection_t *conn;
 	struct link_msg    incoming;
@@ -66,6 +104,7 @@ struct link_call {
 	struct link_writer reply_writer;	/* writes into conn->txbuf */
 	int               reply_consumed;
 	int               error_sent;
+	uid_t             uid;		/* caller, resolved for a broker peer */
 };
 
 /* A parsed AddMatch rule.  Fields are NULL when the rule omits the
@@ -108,6 +147,16 @@ struct link_connection {
 
 	uint32_t            next_serial;
 
+	/* Outbound calls we made on this connection, awaiting replies.
+	 * Only a broker connection uses these today, to ask the bus
+	 * driver who a sender is. */
+	struct {
+		int              used;
+		uint32_t         serial;
+		link_reply_cb_t  cb;
+		void            *userdata;
+	} pending[LINK_PENDING_CAP];
+
 	struct link_server  *server;	/* back-pointer for dispatch */
 };
 
@@ -121,7 +170,8 @@ void __auth_generate_guid(char out[33]);
 int  __auth_client(int fd, uid_t uid);
 
 /* dispatch.c */
-int  __dispatch_message(link_connection_t *conn, const struct link_msg *m);
+int  __dispatch_message(link_connection_t *conn, const struct link_msg *m, size_t framelen);
+void __dispatch_forget_conn(link_connection_t *conn);
 int  __send_error(link_connection_t *conn, const struct link_msg *req,
 		     const char *error_name, const char *text);
 int  __send_method_return(link_connection_t *conn, const struct link_msg *req,
diff --git a/libink/link.h b/libink/link.h
index 5a16e778..afdd5d30 100644
--- a/libink/link.h
+++ b/libink/link.h
@@ -88,6 +88,67 @@ typedef struct {
 	size_t         body_len;
 } link_reply_t;
 
+/* ----------  caller identity on a broker connection  ---------- */
+
+/* Handle for a call parked while its caller is identified.  Opaque,
+ * and safe to hold: it encodes a slot and a generation, so resolving
+ * a stale handle is a no-op rather than a use-after-free. */
+typedef uint64_t link_authz_t;
+
+/* "Nobody asked yet", distinct from any real uid.  link_call_uid()
+ * returns this on a broker connection until something forces the
+ * question, which today only a LINK_METHOD_PRIVILEGED method does. */
+#define LINK_UID_UNKNOWN ((uid_t)-1)
+
+/* A broker sets SENDER to a unique name, ":1.<u32>", so this is very
+ * generous.  Callers that key anything on a sender must reject longer
+ * names rather than truncate: two senders sharing a truncated key
+ * would share an identity. */
+#define LINK_SENDER_MAX 64
+
+/* Answer "which uid is `sender`?" for a privileged call arriving on a
+ * broker connection, where SO_PEERCRED describes the broker and not
+ * the caller.
+ *
+ * Return 0 with *uid set when the answer is already known, 1 to answer
+ * later by calling link_uid_resolved() with `tok`, or -1 when it
+ * cannot be determined, which fails the call closed.  Returning 1
+ * without ever calling link_uid_resolved() leaks the slot and leaves
+ * the caller without a reply, so always answer. */
+typedef int (*link_uid_resolver_t)(link_connection_t *conn, const char *sender,
+				   link_authz_t tok, uid_t *uid, void *userdata);
+
+void link_server_set_uid_resolver(link_server_t *server, link_uid_resolver_t cb,
+				  void *userdata);
+
+/* May `uid` invoke a LINK_METHOD_PRIVILEGED method?  Return non-zero
+ * to allow.  Who counts as privileged is the embedder's policy, not
+ * the library's; with no authorizer installed only uid 0 may. */
+typedef int (*link_authorizer_t)(uid_t uid, void *userdata);
+
+void link_server_set_authorizer(link_server_t *server, link_authorizer_t cb,
+				void *userdata);
+
+/* Complete a deferred resolve and resume the parked call.  Pass
+ * (uid_t)-1 to say the caller could not be identified, which denies
+ * it.  Resolving a handle twice, or one whose connection has since
+ * closed, does nothing. */
+void link_uid_resolved(link_server_t *server, link_authz_t tok, uid_t uid);
+
+/* Called with the reply to an outbound link_connection_call().  `reply`
+ * is NULL if the connection dropped before one arrived. */
+typedef void (*link_reply_cb_t)(link_connection_t *conn, const link_reply_t *reply,
+				void *userdata);
+
+/* Issue a method call on an established connection and invoke `cb`
+ * when the reply lands.  Unlike link_client_call() this never blocks:
+ * the reply is picked up by the normal read loop.  Argument marshalling
+ * matches link_client_call_v(). */
+int link_connection_call(link_connection_t *conn, const char *destination,
+			 const char *path, const char *interface, const char *member,
+			 link_reply_cb_t cb, void *userdata,
+			 const char *signature, ...);
+
 /* ----------  server / connection lifecycle  ---------- */
 
 /* Bind a listening socket at `path` with file mode `mode`, e.g. 0660
diff --git a/libink/marshal.c b/libink/marshal.c
index 48f00363..ecc41166 100644
--- a/libink/marshal.c
+++ b/libink/marshal.c
@@ -376,3 +376,34 @@ int __r_array_begin(struct link_reader *r, size_t *out_end)
 	*out_end = end;
 	return 0;
 }
+
+ssize_t __marshal_va(uint8_t *body, size_t cap, const char *sig, va_list ap)
+{
+	struct link_writer w;
+	const char        *s;
+
+	__w_init(&w, body, cap);
+	for (s = sig; *s; s++) {
+		switch (*s) {
+		case 'y':
+			__w_byte(&w, (uint8_t)va_arg(ap, int));
+			break;
+		case 'b':
+			__w_bool(&w, va_arg(ap, int));
+			break;
+		case 'u':
+			__w_u32(&w, va_arg(ap, uint32_t));
+			break;
+		case 's':
+			__w_string(&w, va_arg(ap, const char *));
+			break;
+		case 'o':
+			__w_path(&w, va_arg(ap, const char *));
+			break;
+		default:
+			return -1;
+		}
+	}
+
+	return __w_finish(&w);
+}
diff --git a/libink/marshal.h b/libink/marshal.h
index 3246d3e7..0a210989 100644
--- a/libink/marshal.h
+++ b/libink/marshal.h
@@ -6,8 +6,10 @@
 #ifndef LIBINK_MARSHAL_H_
 #define LIBINK_MARSHAL_H_
 
+#include <stdarg.h>
 #include <stddef.h>
 #include <stdint.h>
+#include <sys/types.h>
 
 /* struct link_writer is defined in ink.h (public).  Field layout is
  * "opaque" per the public contract; this file's helpers manipulate
@@ -52,4 +54,10 @@ int  __r_align (struct link_reader *r, size_t n);  /* skip to n-byte boundary */
 int  __r_array_begin(struct link_reader *r, size_t *out_end);
 int  __r_done  (const struct link_reader *r);
 
+/* Marshal varargs into `body` (capacity `cap`) according to `sig`.
+ * Returns the marshalled length, or -1 on overflow or an unsupported
+ * type code.  Shared by the synchronous client and the asynchronous
+ * connection-side call. */
+ssize_t __marshal_va(uint8_t *body, size_t cap, const char *sig, va_list ap);
+
 #endif /* LIBINK_MARSHAL_H_ */
diff --git a/libink/proto.c b/libink/proto.c
index 1d7f27c6..96e96cf5 100644
--- a/libink/proto.c
+++ b/libink/proto.c
@@ -389,3 +389,15 @@ size_t __msg_header_size(const struct link_msg *m)
 	/* Generous upper bound used by callers to size send buffers. */
 	return 512;
 }
+
+void __msg_to_reply(link_reply_t *r, const struct link_msg *m)
+{
+	r->type       = m->type;
+	r->signature  = m->signature;
+	r->error_name = m->error_name;
+	r->path       = m->path;
+	r->interface  = m->interface;
+	r->member     = m->member;
+	r->body       = m->body_avail ? m->body : NULL;
+	r->body_len   = m->body_avail;
+}
diff --git a/libink/proto.h b/libink/proto.h
index ef28f4b7..0ea345b1 100644
--- a/libink/proto.h
+++ b/libink/proto.h
@@ -60,6 +60,11 @@ struct link_msg {
  * needed, -1 on malformed input. */
 ssize_t __msg_parse(const uint8_t *buf, size_t len, struct link_msg *out);
 
+/* Project a parsed message onto the public reply view.  Shared by the
+ * synchronous client and the connection-side reply routing so the two
+ * cannot drift as link_reply_t grows. */
+void __msg_to_reply(link_reply_t *r, const struct link_msg *m);
+
 /* Compute the on-wire size of a future message header given the
  * fields we'd populate.  Used to size send buffers. */
 size_t __msg_header_size(const struct link_msg *m);
diff --git a/libink/server.c b/libink/server.c
index dc4b9bff..cb1cd790 100644
--- a/libink/server.c
+++ b/libink/server.c
@@ -159,10 +159,27 @@ int link_server_accept(link_server_t *srv, link_connection_t **out)
 
 	__auth_generate_guid(conn->guid);
 
+
 	*out = conn;
 	return 0;
 }
 
+void link_server_set_uid_resolver(link_server_t *srv, link_uid_resolver_t cb, void *userdata)
+{
+	if (!srv)
+		return;
+	srv->uid_resolver = cb;
+	srv->uid_userdata = userdata;
+}
+
+void link_server_set_authorizer(link_server_t *srv, link_authorizer_t cb, void *userdata)
+{
+	if (!srv)
+		return;
+	srv->authorizer     = cb;
+	srv->authz_userdata = userdata;
+}
+
 link_connection_t *link_server_attach(link_server_t *srv, int fd, uid_t peer_uid,
 				      unsigned int attach_flags)
 {
@@ -199,6 +216,7 @@ link_connection_t *link_server_attach(link_server_t *srv, int fd, uid_t peer_uid
 	conn->broker   = !!(attach_flags & LINK_ATTACH_BROKER);
 	__auth_generate_guid(conn->guid);
 
+
 	return conn;
 
 err_close:
diff --git a/src/dbus.c b/src/dbus.c
index 9f343c8c..c42cf0e5 100644
--- a/src/dbus.c
+++ b/src/dbus.c
@@ -31,6 +31,9 @@
 #ifdef HAVE_DBUS
 
 #include <errno.h>
+#include <grp.h>
+#include <limits.h>
+#include <pwd.h>
 #include <signal.h>
 #include <stdlib.h>
 #include <string.h>
@@ -68,6 +71,7 @@ static size_t             peer_count;
 static struct peer        *sysbus_peer;
 
 static void sysbus_probe(void);
+static void sender_cache_flush(void);
 
 static void peer_drop(struct peer *p)
 {
@@ -79,9 +83,13 @@ static void peer_drop(struct peer *p)
 	peer_count--;
 	free(p);
 
-	/* broker gone; the notify paths probe for its return */
-	if (was_sysbus)
+	/* broker gone; the notify paths probe for its return.  Its unique
+	 * names die with it, so nothing we learned about them is safe to
+	 * carry over to whatever takes its place. */
+	if (was_sysbus) {
 		sysbus_peer = NULL;
+		sender_cache_flush();
+	}
 }
 
 static void peer_cb(uev_t *w, void *arg, int events)
@@ -1171,6 +1179,180 @@ void dbus_notify_condition_change(const char *name, const char *state)
 			 "ConditionChanged", "ss", body, (size_t)blen);
 }
 
+/* ---------- who may change things ----------
+ *
+ * Root, or a member of the group the bus socket is owned by, which is
+ * the same set --with-group already lets connect.  Both gates then say
+ * the same thing, rather than the socket admitting the wheel group and
+ * every method turning it away.
+ *
+ * On the local bus the kernel already made this decision at connect(),
+ * supplementary groups and all, so the lookup only confirms it.  The
+ * system bus has no socket mode to lean on, which is why we ask here
+ * rather than trusting the connection.
+ *
+ * Deliberately uncached: /etc/group changes while Finit runs, and a
+ * privileged call is an operator action, not a hot path.
+ */
+static int caller_is_privileged(uid_t uid, void *userdata)
+{
+	(void)userdata;
+
+	if (uid == 0)
+		return 1;
+
+	/* Group membership comes from NSS, which the C library loads
+	 * with dlopen(), so a build meant to link statically cannot
+	 * count on it.  Compiled out rather than left to fail open:
+	 * root only there, see doc/dbus.md. */
+#ifndef ENABLE_STATIC
+	{
+		gid_t groups[NGROUPS_MAX];
+		int   ngroups = NGROUPS_MAX;
+		struct passwd *pw;
+		int    gid, i;
+
+		gid = getgroup(DEFGROUP);
+		if (gid < 0)
+			return 0;
+
+		pw = getpwuid(uid);
+		if (!pw)
+			return 0;
+
+		if (getgrouplist(pw->pw_name, pw->pw_gid, groups, &ngroups) < 0)
+			return 0;
+
+		for (i = 0; i < ngroups; i++) {
+			if (groups[i] == (gid_t)gid)
+				return 1;
+		}
+	}
+#endif
+
+	return 0;
+}
+
+/* ---------- caller identity on the system bus ----------
+ *
+ * libink parks a privileged call and asks us who sent it; we ask the
+ * bus driver with GetConnectionUnixUser and answer when the reply
+ * lands, through the same event loop as everything else.
+ *
+ * A bus never reuses a unique name, so an answer holds for as long as
+ * that bus runs.  It does not survive the bus restarting, though:
+ * a new dbus-daemon numbers from scratch and :1.7 becomes somebody
+ * else, so peer_drop() empties the cache when the broker goes.
+ *
+ * It is a ring: the oldest entry loses on overflow, and losing one
+ * only costs another round trip.
+ */
+#define SENDER_CACHE_LEN 16
+
+struct sender_uid {
+	char  name[LINK_SENDER_MAX];
+	uid_t uid;
+};
+
+static struct sender_uid sender_cache[SENDER_CACHE_LEN];
+static unsigned          sender_next;
+
+static void sender_cache_flush(void)
+{
+	memset(sender_cache, 0, sizeof(sender_cache));
+	sender_next = 0;
+}
+
+static int sender_cached(const char *sender, uid_t *uid)
+{
+	int i;
+
+	for (i = 0; i < SENDER_CACHE_LEN; i++) {
+		if (sender_cache[i].name[0] && !strcmp(sender_cache[i].name, sender)) {
+			*uid = sender_cache[i].uid;
+			return 1;
+		}
+	}
+
+	return 0;
+}
+
+static void sender_remember(const char *sender, uid_t uid)
+{
+	unsigned i = sender_next++ % SENDER_CACHE_LEN;
+
+	strlcpy(sender_cache[i].name, sender, sizeof(sender_cache[i].name));
+	sender_cache[i].uid = uid;
+}
+
+/* One outstanding GetConnectionUnixUser.  Freed by the reply callback,
+ * which libink guarantees to run exactly once, with a NULL reply if
+ * the connection drops first. */
+struct uid_query {
+	link_authz_t tok;
+	char         sender[LINK_SENDER_MAX];
+};
+
+static void uid_reply_cb(link_connection_t *conn, const link_reply_t *reply, void *userdata)
+{
+	struct uid_query *q = userdata;
+	uid_t             uid = (uid_t)-1;
+	uint32_t          val;
+
+	(void)conn;
+
+	if (!reply) {
+		dbg("connection dropped before %s was identified", q->sender);
+	} else if (reply->type == LINK_MSG_METHOD_RETURN &&
+		   link_reply_get_u32(reply, &val) == 0) {
+		uid = (uid_t)val;
+		sender_remember(q->sender, uid);
+		dbg("sender %s is uid %d", q->sender, (int)uid);
+	} else {
+		dbg("GetConnectionUnixUser(%s) failed: %s", q->sender,
+		    reply->error_name ? reply->error_name : "unexpected reply");
+	}
+
+	link_uid_resolved(server, q->tok, uid);
+	free(q);
+}
+
+static int sysbus_uid_resolver(link_connection_t *conn, const char *sender,
+			       link_authz_t tok, uid_t *uid, void *userdata)
+{
+	struct uid_query *q;
+
+	(void)userdata;
+
+	/* Never truncate: a shortened key could match a different
+	 * sender and hand it someone else's privileges. */
+	if (strlen(sender) >= LINK_SENDER_MAX)
+		return -1;
+
+	if (sender_cached(sender, uid)) {
+		dbg("sender %s is uid %d, from cache", sender, (int)*uid);
+		return 0;
+	}
+
+	dbg("asking the bus driver who %s is ...", sender);
+
+	q = calloc(1, sizeof(*q));
+	if (!q)
+		return -1;
+	q->tok = tok;
+	strlcpy(q->sender, sender, sizeof(q->sender));
+
+	if (link_connection_call(conn, "org.freedesktop.DBus", "/org/freedesktop/DBus",
+				 "org.freedesktop.DBus", "GetConnectionUnixUser",
+				 uid_reply_cb, q, "s", sender) < 0) {
+		dbg("Failed asking the bus driver about %s: %s", sender, strerror(errno));
+		free(q);
+		return -1;
+	}
+
+	return 1;	/* parked; uid_reply_cb() answers */
+}
+
 /* ---------- system-bus attach (opportunistic) ----------
  *
  * If /var/run/dbus/system_bus_socket is reachable, libink connects to
@@ -1179,10 +1361,9 @@ void dbus_notify_condition_change(const char *name, const char *state)
  * peer so the same vtables serve incoming method calls and outgoing
  * signal fan-out reaches the system bus.
  *
- * peer_uid is set to (uid_t)-1 so LINK_METHOD_PRIVILEGED methods
- * reject by default -- per-request sender uid lookup via
- * GetConnectionUnixUser is a follow-up.  Read-only methods
- * (ListServices, Properties.Get, Introspect, ...) work as expected.
+ * peer_uid is (uid_t)-1 because the connection has no single owner;
+ * who is calling is established per message by sysbus_uid_resolver()
+ * below.
  *
  * A bounded SO_SNDTIMEO/SO_RCVTIMEO budget is applied via
  * link_client_open_timeout so a hung dbus-daemon can't stall boot;
@@ -1292,6 +1473,7 @@ static int try_attach_system_bus(uev_ctx_t *ctx)
 	}
 
 	sysbus_peer = p;
+	link_server_set_uid_resolver(server, sysbus_uid_resolver, NULL);
 	sysbus_warned = 0;	/* arm the warning for a later broker restart */
 	logit(LOG_NOTICE, "Registered %s on system bus", FINIT_BUS_NAME);
 	return 0;
@@ -1352,6 +1534,8 @@ int dbus_init(uev_ctx_t *ctx)
 	if (chown(FINIT_BUS_SOCKET, geteuid(), getgroup(DEFGROUP)))
 		err(1, "Failed setting group %s on %s", DEFGROUP, FINIT_BUS_SOCKET);
 
+	link_server_set_authorizer(server, caller_is_privileged, NULL);
+
 	if (link_server_add_object(server, "/org/finit/manager",
 				  &manager_vtable, NULL) < 0) {
 		err(1, "Failed registering Manager1 object");
diff --git a/test/Makefile.am b/test/Makefile.am
index 98723738..a06d9d20 100644
--- a/test/Makefile.am
+++ b/test/Makefile.am
@@ -72,6 +72,7 @@ EXTRA_DIST		+= signal-service.sh
 EXTRA_DIST		+= testserv.sh
 EXTRA_DIST		+= unexpected-restart.sh
 EXTRA_DIST		+= dbus-auth.sh
+EXTRA_DIST		+= dbus-authz.sh
 EXTRA_DIST		+= dbus-bus.sh
 EXTRA_DIST		+= dbus-manager.sh
 EXTRA_DIST		+= dbus-service.sh
@@ -135,6 +136,7 @@ endif
 TESTS			+= unexpected-restart.sh
 if DBUS
 TESTS			+= dbus-auth.sh
+TESTS			+= dbus-authz.sh
 TESTS			+= dbus-bus.sh
 TESTS			+= dbus-manager.sh
 TESTS			+= dbus-service.sh
diff --git a/test/dbus-authz.sh b/test/dbus-authz.sh
new file mode 100755
index 00000000..66badd54
--- /dev/null
+++ b/test/dbus-authz.sh
@@ -0,0 +1,52 @@
+#!/bin/sh
+# Who may invoke a privileged method.  Root always, and anyone in the
+# group the bus socket is owned by, which is DEFGROUP from
+# --with-group, 'root' in a test build.  Everyone else is refused.
+#
+# A caller that gets past the check still has to name a service that
+# exists, so NoSuchService is how we tell "allowed, then failed" apart
+# from "not allowed at all".
+
+set -eu
+
+TEST_DIR=$(dirname "$0")
+
+# shellcheck source=/dev/null
+. "$TEST_DIR/lib/setup.sh"
+# shellcheck source=/dev/null
+. "$TEST_DIR/lib/dbus-setup.sh"
+
+# uid 1000 is 'wheelie', a member of group root in the test sysroot.
+# uid 2 is 'bin', a member of nothing that matters here.
+say "A member of the group may call a privileged method"
+set +e
+allowed=$(texec "$CLIENT" call-s-as-uid 1000 "$BUS" /org/finit/manager \
+	  org.finit.Manager1 Restart nosuchservice 2>&1)
+set -e
+case "$allowed" in
+    *AccessDenied*) fail "Group member was refused: $allowed" ;;
+    *NoSuchService*) assert "Group member passed authorization" 0 -eq 0 ;;
+    *)              fail "Unexpected reply for group member: $allowed" ;;
+esac
+
+say "A caller outside the group may not"
+set +e
+denied=$(texec "$CLIENT" call-s-as-uid 2 "$BUS" /org/finit/manager \
+	 org.finit.Manager1 Restart nosuchservice 2>&1)
+set -e
+case "$denied" in
+    *AccessDenied*)  assert "Non-member refused" 0 -eq 0 ;;
+    *NoSuchService*) fail "Non-member passed authorization: $denied" ;;
+    *)               fail "Unexpected reply for non-member: $denied" ;;
+esac
+
+say "Root is still allowed"
+set +e
+asroot=$(texec "$CLIENT" call-s "$BUS" /org/finit/manager \
+	 org.finit.Manager1 Restart nosuchservice 2>&1)
+set -e
+case "$asroot" in
+    *AccessDenied*) fail "Root was refused: $asroot" ;;
+    *NoSuchService*) assert "Root passed authorization" 0 -eq 0 ;;
+    *)              fail "Unexpected reply for root: $asroot" ;;
+esac
diff --git a/test/skel/etc/group b/test/skel/etc/group
index 696ac862..197977f1 100644
--- a/test/skel/etc/group
+++ b/test/skel/etc/group
@@ -1,4 +1,4 @@
-root:x:0:
+root:x:0:wheelie
 daemon:x:1:
 bin:x:2:
 sys:x:3:
@@ -7,3 +7,4 @@ tty:x:5:
 disk:x:6:
 dialout:x:20:
 nogroup:x:65534:
+wheelie:x:1000:
diff --git a/test/skel/etc/passwd b/test/skel/etc/passwd
index 2933891f..bede8cf6 100644
--- a/test/skel/etc/passwd
+++ b/test/skel/etc/passwd
@@ -3,3 +3,4 @@ daemon:x:1:1:daemon:/usr/sbin:/usr/sbin/nologin
 bin:x:2:2:bin:/bin:/usr/sbin/nologin
 sys:x:3:3:sys:/dev:/usr/sbin/nologin
 nobody:x:65534:65534:nobody:/nonexistent:/usr/sbin/nologin
+wheelie:x:1000:1000:wheelie:/home/wheelie:/bin/sh