Merge pull request #32560 from poettering/varlink-double-fd

varlink: add ability to talk to remote Varlink service binary via ssh

docs/ENVIRONMENT.md

@@ -686,7 +686,8 @@ Tools using the Varlink protocol (such as `varlinkctl`) or sd-bus (such as
 * `$SYSTEMD_VARLINK_LISTEN` – interpreted by some tools that provide a Varlink
   service. Takes a file system path: if specified the tool will listen on an
   `AF_UNIX` stream socket on the specified path in addition to whatever else it
-  would listen on.
+  would listen on. If set to "-" the tool will turn stdin/stdout into a Varlink
+  connection.
 
 `systemd-mountfsd`:
 

man/varlinkctl.xml

@@ -71,16 +71,23 @@
 
     <itemizedlist>
       <listitem><para>A Varlink service reference starting with the <literal>unix:</literal> string, followed
-      by an absolute <constant>AF_UNIX</constant> socket path, or by <literal>@</literal> and an arbitrary string
-      (the latter for referencing sockets in the abstract namespace).</para></listitem>
+      by an absolute <constant>AF_UNIX</constant> socket path, or by <literal>@</literal> and an arbitrary
+      string (the latter for referencing sockets in the abstract namespace). In this case a stream socket
+      connection is made to the specified socket.</para></listitem>
 
       <listitem><para>A Varlink service reference starting with the <literal>exec:</literal> string, followed
-      by an absolute path of a binary to execute.</para></listitem>
+      by an absolute path of a binary to execute. In this case the specified process is forked off locally,
+      with a connected stream socket passed in.</para></listitem>
 
-      <listitem><para>A Varlink service reference starting with the <literal>ssh:</literal> string, followed
+      <listitem><para>A Varlink service reference starting with the <literal>ssh-unix:</literal> string, followed
       by an SSH host specification, followed by <literal>:</literal>, followed by an absolute
       <constant>AF_UNIX</constant> socket path. (This requires OpenSSH 9.4 or newer on the server side,
       abstract namespace sockets are not supported.)</para></listitem>
+
+      <listitem><para>A Varlink service reference starting with the <literal>ssh-exec:</literal> string,
+      followed by an SSH host specification, followed by <literal>:</literal>, followed by a command line. In
+      this case the command is invoked and the Varlink protocol is spoken on the standard input and output of
+      the invoked command.</para></listitem>
     </itemizedlist>
 
     <para>For convenience these two simpler (redundant) service address syntaxes are also supported:</para>
@@ -353,6 +360,22 @@ method Extend(
 {}</programlisting>
     </example>
 
+    <example>
+      <title>Invoking a method remotely via SSH</title>
+
+      <para>The following command acquires a report about the identity of a remote host
+      <literal>somehost</literal> from
+      <citerefentry><refentrytitle>systemd-hostnamed.service</refentrytitle><manvolnum>8</manvolnum></citerefentry>
+      by connecting via SSH to the <constant>AF_UNIX</constant> socket the service listens on:</para>
+
+      <programlisting># varlinkctl call ssh-unix:somehost:/run/systemd/io.systemd.Hostname io.systemd.Hostname.Describe '{}'</programlisting>
+
+      <para>To invoke a Varlink service binary directly on the remote host, rather than talking to a service
+      via <constant>AF_UNIX</constant> can be done like this:</para>
+
+      <programlisting># varlinkctl call ssh-exec:somehost:/usr/bin/systemd-creds org.varlink.service.GetInfo '{}'</programlisting>
+    </example>
+
   </refsect1>
 
   <refsect1>

src/shared/varlink.c

@@ -7,6 +7,7 @@
 
 #include "alloc-util.h"
 #include "errno-util.h"
+#include "escape.h"
 #include "fd-util.h"
 #include "glyph-util.h"
 #include "hashmap.h"
@@ -139,7 +140,8 @@ struct Varlink {
                           * at most. */
         unsigned n_pending;
 
-        int fd;
+        int input_fd;
+        int output_fd;
 
         char *input_buffer; /* valid data starts at input_buffer_index, ends at input_buffer_index+input_buffer_size */
         size_t input_buffer_index;
@@ -185,7 +187,8 @@ struct Varlink {
 
         bool write_disconnected:1;
         bool read_disconnected:1;
-        bool prefer_read_write:1;
+        bool prefer_read:1;
+        bool prefer_write:1;
         bool got_pollhup:1;
 
         bool allow_fd_passing_input:1;
@@ -203,7 +206,8 @@ struct Varlink {
         char *description;
 
         sd_event *event;
-        sd_event_source *io_event_source;
+        sd_event_source *input_event_source;
+        sd_event_source *output_event_source;
         sd_event_source *time_event_source;
         sd_event_source *quit_event_source;
         sd_event_source *defer_event_source;
@@ -357,7 +361,8 @@ static int varlink_new(Varlink **ret) {
 
         *v = (Varlink) {
                 .n_ref = 1,
-                .fd = -EBADF,
+                .input_fd = -EBADF,
+                .output_fd = -EBADF,
 
                 .state = _VARLINK_STATE_INVALID,
 
@@ -387,11 +392,11 @@ int varlink_connect_address(Varlink **ret, const char *address) {
         if (r < 0)
                 return log_debug_errno(r, "Failed to create varlink object: %m");
 
-        v->fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
-        if (v->fd < 0)
+        v->input_fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC|SOCK_NONBLOCK, 0);
+        if (v->input_fd < 0)
                 return log_debug_errno(errno, "Failed to create AF_UNIX socket: %m");
 
-        v->fd = fd_move_above_stdio(v->fd);
+        v->output_fd = v->input_fd = fd_move_above_stdio(v->input_fd);
         v->af = AF_UNIX;
 
         r = sockaddr_un_set_path(&sockaddr.un, address);
@@ -402,9 +407,9 @@ int varlink_connect_address(Varlink **ret, const char *address) {
                 /* This is a file system path, and too long to fit into sockaddr_un. Let's connect via O_PATH
                  * to this socket. */
 
-                r = connect_unix_path(v->fd, AT_FDCWD, address);
+                r = connect_unix_path(v->input_fd, AT_FDCWD, address);
         } else
-                r = RET_NERRNO(connect(v->fd, &sockaddr.sa, r));
+                r = RET_NERRNO(connect(v->input_fd, &sockaddr.sa, r));
 
         if (r < 0) {
                 if (!IN_SET(r, -EAGAIN, -EINPROGRESS))
@@ -507,7 +512,7 @@ int varlink_connect_exec(Varlink **ret, const char *_command, char **_argv) {
         if (r < 0)
                 return log_debug_errno(r, "Failed to create varlink object: %m");
 
-        v->fd = TAKE_FD(pair[0]);
+        v->output_fd = v->input_fd = TAKE_FD(pair[0]);
         v->af = AF_UNIX;
         v->exec_pid = TAKE_PID(pid);
         varlink_set_state(v, VARLINK_IDLE_CLIENT);
@@ -516,7 +521,18 @@ int varlink_connect_exec(Varlink **ret, const char *_command, char **_argv) {
         return 0;
 }
 
-static int varlink_connect_ssh(Varlink **ret, const char *where) {
+static int ssh_path(const char **ret) {
+        assert(ret);
+
+        const char *ssh = secure_getenv("SYSTEMD_SSH") ?: "ssh";
+        if (!path_is_valid(ssh))
+                return log_debug_errno(SYNTHETIC_ERRNO(EINVAL), "SSH path is not valid, refusing: %s", ssh);
+
+        *ret = ssh;
+        return 0;
+}
+
+static int varlink_connect_ssh_unix(Varlink **ret, const char *where) {
         _cleanup_close_pair_ int pair[2] = EBADF_PAIR;
         _cleanup_(sigkill_waitp) pid_t pid = 0;
         int r;
@@ -527,9 +543,10 @@ static int varlink_connect_ssh(Varlink **ret, const char *where) {
         /* Connects to an SSH server via OpenSSH 9.4's -W switch to connect to a remote AF_UNIX socket. For
          * now we do not expose this function directly, but only via varlink_connect_url(). */
 
-        const char *ssh = secure_getenv("SYSTEMD_SSH") ?: "ssh";
-        if (!path_is_valid(ssh))
-                return log_debug_errno(SYNTHETIC_ERRNO(EINVAL), "SSH path is not valid, refusing: %s", ssh);
+        const char *ssh;
+        r = ssh_path(&ssh);
+        if (r < 0)
+                return r;
 
         const char *e = strchr(where, ':');
         if (!e)
@@ -583,7 +600,7 @@ static int varlink_connect_ssh(Varlink **ret, const char *where) {
         if (r < 0)
                 return log_debug_errno(r, "Failed to create varlink object: %m");
 
-        v->fd = TAKE_FD(pair[0]);
+        v->output_fd = v->input_fd = TAKE_FD(pair[0]);
         v->af = AF_UNIX;
         v->exec_pid = TAKE_PID(pid);
         varlink_set_state(v, VARLINK_IDLE_CLIENT);
@@ -592,13 +609,107 @@ static int varlink_connect_ssh(Varlink **ret, const char *where) {
         return 0;
 }
 
+static int varlink_connect_ssh_exec(Varlink **ret, const char *where) {
+        _cleanup_close_pair_ int input_pipe[2] = EBADF_PAIR, output_pipe[2] = EBADF_PAIR;
+        _cleanup_(sigkill_waitp) pid_t pid = 0;
+        int r;
+
+        assert_return(ret, -EINVAL);
+        assert_return(where, -EINVAL);
+
+        /* Connects to an SSH server to connect to a remote process' stdin/stdout. For now we do not expose
+         * this function directly, but only via varlink_connect_url(). */
+
+        const char *ssh;
+        r = ssh_path(&ssh);
+        if (r < 0)
+                return r;
+
+        const char *e = strchr(where, ':');
+        if (!e)
+                return log_debug_errno(SYNTHETIC_ERRNO(EINVAL), "SSH specification lacks a : separator between host and path, refusing: %s", where);
+
+        _cleanup_free_ char *h = strndup(where, e - where);
+        if (!h)
+                return log_oom_debug();
+
+        _cleanup_strv_free_ char **cmdline = NULL;
+        r = strv_split_full(&cmdline, e + 1, /* separators= */ NULL, EXTRACT_CUNESCAPE|EXTRACT_UNQUOTE);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to split command line: %m");
+        if (strv_isempty(cmdline))
+                return log_debug_errno(SYNTHETIC_ERRNO(EINVAL), "Remote command line is empty, refusing.");
+
+        _cleanup_strv_free_ char **full_cmdline = NULL;
+        full_cmdline = strv_new("ssh", "-e", "none", "-T", h, "env", "SYSTEMD_VARLINK_LISTEN=-");
+        if (!full_cmdline)
+                return log_oom_debug();
+        r = strv_extend_strv(&full_cmdline, cmdline, /* filter_duplicates= */ false);
+        if (r < 0)
+                return log_oom_debug();
+
+        _cleanup_free_ char *j = NULL;
+        j = quote_command_line(full_cmdline, SHELL_ESCAPE_EMPTY);
+        if (!j)
+                return log_oom_debug();
+
+        log_debug("Forking off SSH child process: %s", j);
+
+        if (pipe2(input_pipe, O_CLOEXEC) < 0)
+                return log_debug_errno(errno, "Failed to allocate input pipe: %m");
+        if (pipe2(output_pipe, O_CLOEXEC) < 0)
+                return log_debug_errno(errno, "Failed to allocate output pipe: %m");
+
+        r = safe_fork_full(
+                        "(sd-vlssh)",
+                        /* stdio_fds= */ (int[]) { input_pipe[0], output_pipe[1], STDERR_FILENO },
+                        /* except_fds= */ NULL,
+                        /* n_except_fds= */ 0,
+                        FORK_RESET_SIGNALS|FORK_CLOSE_ALL_FDS|FORK_DEATHSIG_SIGTERM|FORK_REOPEN_LOG|FORK_LOG|FORK_RLIMIT_NOFILE_SAFE|FORK_REARRANGE_STDIO,
+                        &pid);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to spawn process: %m");
+        if (r == 0) {
+                /* Child */
+                execvp(ssh, full_cmdline);
+                log_debug_errno(errno, "Failed to invoke %s: %m", j);
+                _exit(EXIT_FAILURE);
+        }
+
+        input_pipe[0] = safe_close(input_pipe[0]);
+        output_pipe[1] = safe_close(output_pipe[1]);
+
+        r = fd_nonblock(input_pipe[1], true);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to make input pipe non-blocking: %m");
+
+        r = fd_nonblock(output_pipe[0], true);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to make output pipe non-blocking: %m");
+
+        Varlink *v;
+        r = varlink_new(&v);
+        if (r < 0)
+                return log_debug_errno(r, "Failed to create varlink object: %m");
+
+        v->input_fd = TAKE_FD(output_pipe[0]);
+        v->output_fd = TAKE_FD(input_pipe[1]);
+        v->af = AF_UNSPEC;
+        v->exec_pid = TAKE_PID(pid);
+        varlink_set_state(v, VARLINK_IDLE_CLIENT);
+
+        *ret = v;
+        return 0;
+}
+
 int varlink_connect_url(Varlink **ret, const char *url) {
         _cleanup_free_ char *c = NULL;
         const char *p;
         enum {
                 SCHEME_UNIX,
                 SCHEME_EXEC,
-                SCHEME_SSH,
+                SCHEME_SSH_UNIX,
+                SCHEME_SSH_EXEC,
         } scheme;
         int r;
 
@@ -616,8 +727,10 @@ int varlink_connect_url(Varlink **ret, const char *url) {
                 scheme = SCHEME_UNIX;
         else if ((p = startswith(url, "exec:")))
                 scheme = SCHEME_EXEC;
-        else if ((p = startswith(url, "ssh:")))
-                scheme = SCHEME_SSH;
+        else if ((p = STARTSWITH_SET(url, "ssh:", "ssh-unix:")))
+                scheme = SCHEME_SSH_UNIX;
+        else if ((p = startswith(url, "ssh-exec:")))
+                scheme = SCHEME_SSH_EXEC;
         else
                 return log_debug_errno(SYNTHETIC_ERRNO(EPROTONOSUPPORT), "URL scheme not supported.");
 
@@ -626,8 +739,10 @@ int varlink_connect_url(Varlink **ret, const char *url) {
         if (p[strcspn(p, ";?#")] != '\0')
                 return log_debug_errno(SYNTHETIC_ERRNO(EPROTONOSUPPORT), "URL parameterization with ';', '?', '#' not supported.");
 
-        if (scheme == SCHEME_SSH)
-                return varlink_connect_ssh(ret, p);
+        if (scheme == SCHEME_SSH_UNIX)
+                return varlink_connect_ssh_unix(ret, p);
+        if (scheme == SCHEME_SSH_EXEC)
+                return varlink_connect_ssh_exec(ret, p);
 
         if (scheme == SCHEME_EXEC || p[0] != '@') { /* no path validity checks for abstract namespace sockets */
 
@@ -648,23 +763,37 @@ int varlink_connect_url(Varlink **ret, const char *url) {
         return varlink_connect_address(ret, c ?: p);
 }
 
-int varlink_connect_fd(Varlink **ret, int fd) {
+int varlink_connect_fd_pair(Varlink **ret, int input_fd, int output_fd, const struct ucred *override_ucred) {
         Varlink *v;
         int r;
 
         assert_return(ret, -EINVAL);
-        assert_return(fd >= 0, -EBADF);
+        assert_return(input_fd >= 0, -EBADF);
+        assert_return(output_fd >= 0, -EBADF);
 
-        r = fd_nonblock(fd, true);
+        r = fd_nonblock(input_fd, true);
         if (r < 0)
-                return log_debug_errno(r, "Failed to make fd %d nonblocking: %m", fd);
+                return log_debug_errno(r, "Failed to make input fd %d nonblocking: %m", input_fd);
+
+        if (input_fd != output_fd) {
+                r = fd_nonblock(output_fd, true);
+                if (r < 0)
+                        return log_debug_errno(r, "Failed to make output fd %d nonblocking: %m", output_fd);
+        }
 
         r = varlink_new(&v);
         if (r < 0)
                 return log_debug_errno(r, "Failed to create varlink object: %m");
 
-        v->fd = fd;
-        v->af = -1,
+        v->input_fd = input_fd;
+        v->output_fd = output_fd;
+        v->af = -1;
+
+        if (override_ucred) {
+                v->ucred = *override_ucred;
+                v->ucred_acquired = true;
+        }
+
         varlink_set_state(v, VARLINK_IDLE_CLIENT);
 
         /* Note that if this function is called we assume the passed socket (if it is one) is already
@@ -678,10 +807,15 @@ int varlink_connect_fd(Varlink **ret, int fd) {
         return 0;
 }
 
+int varlink_connect_fd(Varlink **ret, int fd) {
+        return varlink_connect_fd_pair(ret, fd, fd, /* override_ucred= */ NULL);
+}
+
 static void varlink_detach_event_sources(Varlink *v) {
         assert(v);
 
-        v->io_event_source = sd_event_source_disable_unref(v->io_event_source);
+        v->input_event_source = sd_event_source_disable_unref(v->input_event_source);
+        v->output_event_source = sd_event_source_disable_unref(v->output_event_source);
         v->time_event_source = sd_event_source_disable_unref(v->time_event_source);
         v->quit_event_source = sd_event_source_disable_unref(v->quit_event_source);
         v->defer_event_source = sd_event_source_disable_unref(v->defer_event_source);
@@ -706,7 +840,11 @@ static void varlink_clear(Varlink *v) {
 
         varlink_detach_event_sources(v);
 
-        v->fd = safe_close(v->fd);
+        if (v->input_fd != v->output_fd) {
+                v->input_fd = safe_close(v->input_fd);
+                v->output_fd = safe_close(v->output_fd);
+        } else
+                v->output_fd = v->input_fd = safe_close(v->input_fd);
 
         varlink_clear_current(v);
 
@@ -821,7 +959,7 @@ static int varlink_write(Varlink *v) {
         if (v->output_buffer_size == 0)
                 return 0;
 
-        assert(v->fd >= 0);
+        assert(v->output_fd >= 0);
 
         if (v->n_output_fds > 0) { /* If we shall send fds along, we must use sendmsg() */
                 struct iovec iov = {
@@ -842,20 +980,20 @@ static int varlink_write(Varlink *v) {
                 control->cmsg_type = SCM_RIGHTS;
                 memcpy(CMSG_DATA(control), v->output_fds, sizeof(int) * v->n_output_fds);
 
-                n = sendmsg(v->fd, &mh, MSG_DONTWAIT|MSG_NOSIGNAL);
+                n = sendmsg(v->output_fd, &mh, MSG_DONTWAIT|MSG_NOSIGNAL);
         } else {
                 /* We generally prefer recv()/send() (mostly because of MSG_NOSIGNAL) but also want to be compatible
                  * with non-socket IO, hence fall back automatically.
                  *
                  * Use a local variable to help gcc figure out that we set 'n' in all cases. */
-                bool prefer_write = v->prefer_read_write;
+                bool prefer_write = v->prefer_write;
                 if (!prefer_write) {
-                        n = send(v->fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size, MSG_DONTWAIT|MSG_NOSIGNAL);
+                        n = send(v->output_fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size, MSG_DONTWAIT|MSG_NOSIGNAL);
                         if (n < 0 && errno == ENOTSOCK)
-                                prefer_write = v->prefer_read_write = true;
+                                prefer_write = v->prefer_write = true;
                 }
                 if (prefer_write)
-                        n = write(v->fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size);
+                        n = write(v->output_fd, v->output_buffer + v->output_buffer_index, v->output_buffer_size);
         }
         if (n < 0) {
                 if (errno == EAGAIN)
@@ -914,7 +1052,7 @@ static int varlink_read(Varlink *v) {
         if (v->input_buffer_size >= VARLINK_BUFFER_MAX)
                 return -ENOBUFS;
 
-        assert(v->fd >= 0);
+        assert(v->input_fd >= 0);
 
         if (MALLOC_SIZEOF_SAFE(v->input_buffer) <= v->input_buffer_index + v->input_buffer_size) {
                 size_t add;
@@ -961,16 +1099,16 @@ static int varlink_read(Varlink *v) {
                         .msg_controllen = v->input_control_buffer_size,
                 };
 
-                n = recvmsg_safe(v->fd, &mh, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
+                n = recvmsg_safe(v->input_fd, &mh, MSG_DONTWAIT|MSG_CMSG_CLOEXEC);
         } else {
-                bool prefer_read = v->prefer_read_write;
+                bool prefer_read = v->prefer_read;
                 if (!prefer_read) {
-                        n = recv(v->fd, p, rs, MSG_DONTWAIT);
+                        n = recv(v->input_fd, p, rs, MSG_DONTWAIT);
                         if (n < 0 && errno == ENOTSOCK)
-                                prefer_read = v->prefer_read_write = true;
+                                prefer_read = v->prefer_read = true;
                 }
                 if (prefer_read)
-                        n = read(v->fd, p, rs);
+                        n = read(v->input_fd, p, rs);
         }
         if (n < 0) {
                 if (errno == EAGAIN)
@@ -1666,7 +1804,7 @@ static void handle_revents(Varlink *v, int revents) {
 }
 
 int varlink_wait(Varlink *v, usec_t timeout) {
-        int r, fd, events;
+        int r, events;
         usec_t t;
 
         assert_return(v, -EINVAL);
@@ -1691,22 +1829,42 @@ int varlink_wait(Varlink *v, usec_t timeout) {
             (t == USEC_INFINITY || timeout < t))
                 t = timeout;
 
-        fd = varlink_get_fd(v);
-        if (fd < 0)
-                return fd;
-
         events = varlink_get_events(v);
         if (events < 0)
                 return events;
 
-        r = fd_wait_for_event(fd, events, t);
+        struct pollfd pollfd[2];
+        size_t n_poll_fd = 0;
+
+        if (v->input_fd == v->output_fd) {
+                pollfd[n_poll_fd++] = (struct pollfd) {
+                        .fd = v->input_fd,
+                        .events = events,
+                };
+        } else {
+                pollfd[n_poll_fd++] = (struct pollfd) {
+                        .fd = v->input_fd,
+                        .events = events & POLLIN,
+                };
+                pollfd[n_poll_fd++] = (struct pollfd) {
+                        .fd = v->output_fd,
+                        .events = events & POLLOUT,
+                };
+        };
+
+        r = ppoll_usec(pollfd, n_poll_fd, t);
         if (ERRNO_IS_NEG_TRANSIENT(r)) /* Treat EINTR as not a timeout, but also nothing happened, and
                                         * the caller gets a chance to call back into us */
                 return 1;
         if (r <= 0)
                 return r;
 
-        handle_revents(v, r);
+        /* Merge the seen events into one */
+        int revents = 0;
+        FOREACH_ARRAY(p, pollfd, n_poll_fd)
+                revents |= p->revents;
+
+        handle_revents(v, revents);
         return 1;
 }
 
@@ -1725,10 +1883,12 @@ int varlink_get_fd(Varlink *v) {
 
         if (v->state == VARLINK_DISCONNECTED)
                 return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
-        if (v->fd < 0)
+        if (v->input_fd != v->output_fd)
+                return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADF), "Separate file descriptors for input/output set.");
+        if (v->input_fd < 0)
                 return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADF), "No valid fd.");
 
-        return v->fd;
+        return v->input_fd;
 }
 
 int varlink_get_events(Varlink *v) {
@@ -1797,7 +1957,7 @@ int varlink_flush(Varlink *v) {
                         continue;
                 }
 
-                r = fd_wait_for_event(v->fd, POLLOUT, USEC_INFINITY);
+                r = fd_wait_for_event(v->output_fd, POLLOUT, USEC_INFINITY);
                 if (ERRNO_IS_NEG_TRANSIENT(r))
                         continue;
                 if (r < 0)
@@ -2794,7 +2954,12 @@ static int varlink_acquire_ucred(Varlink *v) {
         if (v->ucred_acquired)
                 return 0;
 
-        r = getpeercred(v->fd, &v->ucred);
+        /* If we are connected asymmetrically, let's refuse, since it's not clear if caller wants to know
+         * peer on read or write fd */
+        if (v->input_fd != v->output_fd)
+                return -EBADF;
+
+        r = getpeercred(v->input_fd, &v->ucred);
         if (r < 0)
                 return r;
 
@@ -2859,7 +3024,10 @@ static int varlink_acquire_pidfd(Varlink *v) {
         if (v->peer_pidfd >= 0)
                 return 0;
 
-        v->peer_pidfd = getpeerpidfd(v->fd);
+        if (v->input_fd != v->output_fd)
+                return -EBADF;
+
+        v->peer_pidfd = getpeerpidfd(v->input_fd);
         if (v->peer_pidfd < 0)
                 return v->peer_pidfd;
 
@@ -2962,7 +3130,14 @@ static int prepare_callback(sd_event_source *s, void *userdata) {
         if (e < 0)
                 return e;
 
-        r = sd_event_source_set_io_events(v->io_event_source, e);
+        if (v->input_event_source == v->output_event_source)
+                /* Same fd for input + output */
+                r = sd_event_source_set_io_events(v->input_event_source, e);
+        else {
+                r = sd_event_source_set_io_events(v->input_event_source, e & EPOLLIN);
+                if (r >= 0)
+                        r = sd_event_source_set_io_events(v->output_event_source, e & EPOLLOUT);
+        }
         if (r < 0)
                 return varlink_log_errno(v, r, "Failed to set source events: %m");
 
@@ -3029,19 +3204,33 @@ int varlink_attach_event(Varlink *v, sd_event *e, int64_t priority) {
 
         (void) sd_event_source_set_description(v->quit_event_source, "varlink-quit");
 
-        r = sd_event_add_io(v->event, &v->io_event_source, v->fd, 0, io_callback, v);
+        r = sd_event_add_io(v->event, &v->input_event_source, v->input_fd, 0, io_callback, v);
         if (r < 0)
                 goto fail;
 
-        r = sd_event_source_set_prepare(v->io_event_source, prepare_callback);
+        r = sd_event_source_set_prepare(v->input_event_source, prepare_callback);
         if (r < 0)
                 goto fail;
 
-        r = sd_event_source_set_priority(v->io_event_source, priority);
+        r = sd_event_source_set_priority(v->input_event_source, priority);
         if (r < 0)
                 goto fail;
 
-        (void) sd_event_source_set_description(v->io_event_source, "varlink-io");
+        (void) sd_event_source_set_description(v->input_event_source, "varlink-input");
+
+        if (v->input_fd == v->output_fd)
+                v->output_event_source = sd_event_source_ref(v->input_event_source);
+        else {
+                r = sd_event_add_io(v->event, &v->output_event_source, v->output_fd, 0, io_callback, v);
+                if (r < 0)
+                        goto fail;
+
+                r = sd_event_source_set_priority(v->output_event_source, priority);
+                if (r < 0)
+                        goto fail;
+
+                (void) sd_event_source_set_description(v->output_event_source, "varlink-output");
+        }
 
         r = sd_event_add_defer(v->event, &v->defer_event_source, defer_callback, v);
         if (r < 0)
@@ -3187,16 +3376,23 @@ static int verify_unix_socket(Varlink *v) {
          *    • otherwise: v->af contains the address family we determined */
 
         if (v->af < 0) {
+                /* If we have distinct input + output fds, we don't consider ourselves to be connected via a regular
+                 * AF_UNIX socket. */
+                if (v->input_fd != v->output_fd) {
+                        v->af = AF_UNSPEC;
+                        return -ENOTSOCK;
+                }
+
                 struct stat st;
 
-                if (fstat(v->fd, &st) < 0)
+                if (fstat(v->input_fd, &st) < 0)
                         return -errno;
                 if (!S_ISSOCK(st.st_mode)) {
                         v->af = AF_UNSPEC;
                         return -ENOTSOCK;
                 }
 
-                v->af = socket_get_family(v->fd);
+                v->af = socket_get_family(v->input_fd);
                 if (v->af < 0)
                         return v->af;
         }
@@ -3379,19 +3575,34 @@ static int count_connection(VarlinkServer *server, const struct ucred *ucred) {
         return 0;
 }
 
-int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret) {
+int varlink_server_add_connection_pair(
+                VarlinkServer *server,
+                int input_fd,
+                int output_fd,
+                const struct ucred *override_ucred,
+                Varlink **ret) {
+
         _cleanup_(varlink_unrefp) Varlink *v = NULL;
         struct ucred ucred = UCRED_INVALID;
         bool ucred_acquired;
         int r;
 
         assert_return(server, -EINVAL);
-        assert_return(fd >= 0, -EBADF);
+        assert_return(input_fd >= 0, -EBADF);
+        assert_return(output_fd >= 0, -EBADF);
 
         if ((server->flags & (VARLINK_SERVER_ROOT_ONLY|VARLINK_SERVER_ACCOUNT_UID)) != 0) {
-                r = getpeercred(fd, &ucred);
-                if (r < 0)
-                        return varlink_server_log_errno(server, r, "Failed to acquire peer credentials of incoming socket, refusing: %m");
+
+                if (override_ucred)
+                        ucred = *override_ucred;
+                else {
+                        if (input_fd != output_fd)
+                                return varlink_server_log_errno(server, SYNTHETIC_ERRNO(EOPNOTSUPP), "Cannot determine peer identity of connection with separate input/output, refusing: %m");
+
+                        r = getpeercred(input_fd, &ucred);
+                        if (r < 0)
+                                return varlink_server_log_errno(server, r, "Failed to acquire peer credentials of incoming socket, refusing: %m");
+                }
 
                 ucred_acquired = true;
 
@@ -3411,7 +3622,8 @@ int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret)
         if (r < 0)
                 return r;
 
-        v->fd = fd;
+        v->input_fd = input_fd;
+        v->output_fd = output_fd;
         if (server->flags & VARLINK_SERVER_INHERIT_USERDATA)
                 v->userdata = server->userdata;
 
@@ -3421,7 +3633,7 @@ int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret)
         }
 
         _cleanup_free_ char *desc = NULL;
-        if (asprintf(&desc, "%s-%i", varlink_server_description(server), v->fd) >= 0)
+        if (asprintf(&desc, "%s-%i-%i", varlink_server_description(server), input_fd, output_fd) >= 0)
                 v->description = TAKE_PTR(desc);
 
         /* Link up the server and the connection, and take reference in both directions. Note that the
@@ -3436,7 +3648,8 @@ int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret)
                 r = varlink_attach_event(v, server->event, server->event_priority);
                 if (r < 0) {
                         varlink_log_errno(v, r, "Failed to attach new connection: %m");
-                        v->fd = -EBADF; /* take the fd out of the connection again */
+                        TAKE_FD(v->input_fd); /* take the fd out of the connection again */
+                        TAKE_FD(v->output_fd);
                         varlink_close(v);
                         return r;
                 }
@@ -3448,6 +3661,10 @@ int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret)
         return 0;
 }
 
+int varlink_server_add_connection(VarlinkServer *server, int fd, Varlink **ret) {
+        return varlink_server_add_connection_pair(server, fd, fd, /* override_ucred= */ NULL, ret);
+}
+
 static VarlinkServerSocket *varlink_server_socket_free(VarlinkServerSocket *ss) {
         if (!ss)
                 return NULL;
@@ -3597,6 +3814,66 @@ int varlink_server_listen_address(VarlinkServer *s, const char *address, mode_t
         return 0;
 }
 
+int varlink_server_add_connection_stdio(VarlinkServer *s, Varlink **ret) {
+        _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF;
+        int r;
+
+        assert_return(s, -EINVAL);
+
+        input_fd = fcntl(STDIN_FILENO, F_DUPFD_CLOEXEC, 3);
+        if (input_fd < 0)
+                return -errno;
+
+        output_fd = fcntl(STDOUT_FILENO, F_DUPFD_CLOEXEC, 3);
+        if (output_fd < 0)
+                return -errno;
+
+        r = rearrange_stdio(-EBADF, -EBADF, STDERR_FILENO);
+        if (r < 0)
+                return r;
+
+        r = fd_nonblock(input_fd, true);
+        if (r < 0)
+                return r;
+
+        r = fd_nonblock(output_fd, true);
+        if (r < 0)
+                return r;
+
+        struct stat input_st;
+        if (fstat(input_fd, &input_st) < 0)
+                return -errno;
+
+        struct stat output_st;
+        if (fstat(output_fd, &output_st) < 0)
+                return -errno;
+
+        /* If stdin/stdout are both pipes and have the same owning uid/gid then let's synthesize a "struct
+         * ucred" from the owning UID/GID, since we got them passed in with such ownership. We'll not fill in
+         * the PID however, since there's no way to know which process created a pipe. */
+        struct ucred ucred, *pucred;
+        if (S_ISFIFO(input_st.st_mode) &&
+            S_ISFIFO(output_st.st_mode) &&
+            input_st.st_uid == output_st.st_uid &&
+            input_st.st_gid == output_st.st_gid) {
+                ucred = (struct ucred) {
+                        .uid = input_st.st_uid,
+                        .gid = input_st.st_gid,
+                };
+                pucred = &ucred;
+        } else
+                pucred = NULL;
+
+        r = varlink_server_add_connection_pair(s, input_fd, output_fd, pucred, ret);
+        if (r < 0)
+                return r;
+
+        TAKE_FD(input_fd);
+        TAKE_FD(output_fd);
+
+        return 0;
+}
+
 int varlink_server_listen_auto(VarlinkServer *s) {
         _cleanup_strv_free_ char **names = NULL;
         int r, n = 0;
@@ -3637,12 +3914,17 @@ int varlink_server_listen_auto(VarlinkServer *s) {
                 n++;
         }
 
-        /* For debug purposes let's listen on an explicitly specified address */
+        /* Let's listen on an explicitly specified address */
         const char *e = secure_getenv("SYSTEMD_VARLINK_LISTEN");
         if (e) {
-                r = varlink_server_listen_address(s, e, FLAGS_SET(s->flags, VARLINK_SERVER_ROOT_ONLY) ? 0600 : 0666);
+                if (streq(e, "-"))
+                        r = varlink_server_add_connection_stdio(s, /* ret= */ NULL);
+                else
+                        r = varlink_server_listen_address(s, e, FLAGS_SET(s->flags, VARLINK_SERVER_ROOT_ONLY) ? 0600 : 0666);
                 if (r < 0)
                         return r;
+
+                n++;
         }
 
         return n;
@@ -4092,7 +4374,7 @@ int varlink_invocation(VarlinkInvocationFlags flags) {
 
         /* Returns true if this is a "pure" varlink server invocation, i.e. with one fd passed. */
 
-        const char *e = secure_getenv("SYSTEMD_VARLINK_LISTEN"); /* Permit a manual override for testing purposes */
+        const char *e = secure_getenv("SYSTEMD_VARLINK_LISTEN"); /* Permit an explicit override */
         if (e)
                 return true;
 

src/shared/varlink.h

@@ -1,6 +1,8 @@
 /* SPDX-License-Identifier: LGPL-2.1-or-later */
 #pragma once
 
+#include <sys/socket.h>
+
 #include "sd-event.h"
 #include "sd-json.h"
 
@@ -61,6 +63,7 @@ int varlink_connect_address(Varlink **ret, const char *address);
 int varlink_connect_exec(Varlink **ret, const char *command, char **argv);
 int varlink_connect_url(Varlink **ret, const char *url);
 int varlink_connect_fd(Varlink **ret, int fd);
+int varlink_connect_fd_pair(Varlink **ret, int input_fd, int output_fd, const struct ucred *override_ucred);
 
 Varlink* varlink_ref(Varlink *link);
 Varlink* varlink_unref(Varlink *v);
@@ -218,6 +221,8 @@ int varlink_server_listen_address(VarlinkServer *s, const char *address, mode_t
 int varlink_server_listen_fd(VarlinkServer *s, int fd);
 int varlink_server_listen_auto(VarlinkServer *s);
 int varlink_server_add_connection(VarlinkServer *s, int fd, Varlink **ret);
+int varlink_server_add_connection_pair(VarlinkServer *s, int input_fd, int output_fd, const struct ucred *ucred_override, Varlink **ret);
+int varlink_server_add_connection_stdio(VarlinkServer *s, Varlink **ret);
 
 /* Bind callbacks */
 int varlink_server_bind_method(VarlinkServer *s, const char *method, VarlinkMethod callback);

test/units/TEST-74-AUX-UTILS.varlinkctl.sh

@@ -75,7 +75,7 @@ rm_rf_sshbindir() {
 
 trap rm_rf_sshbindir EXIT
 
-# Create a fake "ssh" binary that validates everything works as expected
+# Create a fake "ssh" binary that validates everything works as expected if invoked for the "ssh-unix:" Varlink transport
 cat > "$SSHBINDIR"/ssh <<'EOF'
 #!/bin/sh
 
@@ -87,10 +87,29 @@ test "$3" = "foobar"
 
 exec socat - UNIX-CONNECT:/run/systemd/journal/io.systemd.journal
 EOF
-
 chmod +x "$SSHBINDIR"/ssh
 
-SYSTEMD_SSH="$SSHBINDIR/ssh" varlinkctl info ssh:foobar:/run/systemd/journal/io.systemd.journal
+SYSTEMD_SSH="$SSHBINDIR/ssh" varlinkctl info ssh-unix:foobar:/run/systemd/journal/io.systemd.journal
+
+# Now build another fake "ssh" binary that does the same for "ssh-exec:"
+cat > "$SSHBINDIR"/ssh <<'EOF'
+#!/bin/sh
+
+set -xe
+
+test "$1" = "-e"
+test "$2" = "none"
+test "$3" = "-T"
+test "$4" = "foobar"
+test "$5" = "env"
+test "$6" = "SYSTEMD_VARLINK_LISTEN=-"
+test "$7" = "systemd-sysext"
+
+SYSTEMD_VARLINK_LISTEN=- exec systemd-sysext
+EOF
+chmod +x "$SSHBINDIR"/ssh
+
+SYSTEMD_SSH="$SSHBINDIR/ssh" varlinkctl info ssh-exec:foobar:systemd-sysext
 
 # Go through all varlink sockets we can find under /run/systemd/ for some extra coverage
 find /run/systemd/ -name "io.systemd*" -type s | while read -r socket; do