diff --git a/man/varlinkctl.xml b/man/varlinkctl.xml
index b96e35372f..6aff5a05e1 100644
--- a/man/varlinkctl.xml
+++ b/man/varlinkctl.xml
@@ -269,6 +269,30 @@
+
+
+
+ When used with call: request a protocol upgrade. The method call
+ is sent with the upgrade flag set. The service is expected to send a single
+ reply confirming the upgrade. After the reply, the Varlink protocol is no longer in effect on
+ the connection.
+
+ If is not specified, varlinkctl acts as a
+ bidirectional proxy: data read from standard input is forwarded to the upgraded connection, and data
+ received from the connection is written to standard output.
+
+ If is specified, the upgraded connection socket is placed on both
+ standard input and standard output of the invoked process. This is similar to the regular
+ behavior (without ), which places the method call
+ reply on standard input. The invoked process can thus simply read from and write to
+ stdin/stdout to communicate over the upgraded protocol.
+
+ This option may not be combined with , ,
+ , , or .
+
+
+
+
diff --git a/shell-completion/bash/varlinkctl b/shell-completion/bash/varlinkctl
index e6557ffb02..53c62dbae5 100644
--- a/shell-completion/bash/varlinkctl
+++ b/shell-completion/bash/varlinkctl
@@ -43,7 +43,7 @@ _varlinkctl() {
local cur=${COMP_WORDS[COMP_CWORD]} prev=${COMP_WORDS[COMP_CWORD-1]}
local -A OPTS=(
[STANDALONE]='-h --help --version --no-pager -q --quiet --no-ask-password
- --oneway --collect --more --exec -j -E'
+ --oneway --collect --more --exec --upgrade -j -E'
[ARG]='--graceful --timeout --push-fd --json'
)
diff --git a/shell-completion/zsh/_varlinkctl b/shell-completion/zsh/_varlinkctl
index f33eea24b6..a071f6ec2e 100644
--- a/shell-completion/zsh/_varlinkctl
+++ b/shell-completion/zsh/_varlinkctl
@@ -65,6 +65,7 @@ local -a opts=(
'--no-pager[Do not pipe output to a pager]'
'--more[Request multiple responses]'
'--collect[Collect multiple responses in a JSON array]'
+ '--upgrade[Upgrade connection to raw protocol after method call]'
{-j+,--json=}'[Output as json]:json-mode:(pretty short)'
)
_arguments -S $opts '*:: := _varlinkctl_command'
diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym
index e90a9460e2..6af86aa2b4 100644
--- a/src/libsystemd/libsystemd.sym
+++ b/src/libsystemd/libsystemd.sym
@@ -1093,5 +1093,6 @@ global:
LIBSYSTEMD_261 {
global:
+ sd_varlink_call_and_upgrade;
sd_varlink_set_sentinel;
} LIBSYSTEMD_260;
diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c
index 9938ca7063..64606adcea 100644
--- a/src/libsystemd/sd-varlink/sd-varlink.c
+++ b/src/libsystemd/sd-varlink/sd-varlink.c
@@ -894,7 +894,12 @@ static int varlink_read(sd_varlink *v) {
}
p = v->input_buffer + v->input_buffer_index + v->input_buffer_size;
- rs = MALLOC_SIZEOF_SAFE(v->input_buffer) - (v->input_buffer_index + v->input_buffer_size);
+
+ /* When a protocol upgrade is requested we can't consume any post-upgrade data from the socket buffer */
+ if (v->protocol_upgrade)
+ rs = 1;
+ else
+ rs = MALLOC_SIZEOF_SAFE(v->input_buffer) - (v->input_buffer_index + v->input_buffer_size);
if (v->allow_fd_passing_input > 0) {
iov = IOVEC_MAKE(p, rs);
@@ -1493,7 +1498,7 @@ static int varlink_dispatch_method(sd_varlink *v) {
} else if (streq(k, "oneway")) {
- if ((flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE)) != 0)
+ if ((flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE|SD_VARLINK_METHOD_UPGRADE)) != 0)
goto invalid;
if (!sd_json_variant_is_boolean(e))
@@ -1504,7 +1509,7 @@ static int varlink_dispatch_method(sd_varlink *v) {
} else if (streq(k, "more")) {
- if ((flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE)) != 0)
+ if ((flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE|SD_VARLINK_METHOD_UPGRADE)) != 0)
goto invalid;
if (!sd_json_variant_is_boolean(e))
@@ -1513,6 +1518,17 @@ static int varlink_dispatch_method(sd_varlink *v) {
if (sd_json_variant_boolean(e))
flags |= SD_VARLINK_METHOD_MORE;
+ } else if (streq(k, "upgrade")) {
+
+ if ((flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE|SD_VARLINK_METHOD_UPGRADE)) != 0)
+ goto invalid;
+
+ if (!sd_json_variant_is_boolean(e))
+ goto invalid;
+
+ if (sd_json_variant_boolean(e))
+ flags |= SD_VARLINK_METHOD_UPGRADE;
+
} else
goto invalid;
}
@@ -2256,19 +2272,13 @@ _public_ int sd_varlink_observeb(sd_varlink *v, const char *method, ...) {
return sd_varlink_observe(v, method, parameters);
}
-_public_ int sd_varlink_call_full(
- sd_varlink *v,
- const char *method,
- sd_json_variant *parameters,
- sd_json_variant **ret_parameters,
- const char **ret_error_id,
- sd_varlink_reply_flags_t *ret_flags) {
-
- _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
+/* On success v->state will equal VARLINK_CALLED, the caller is responsible to adjust the state further if
+ * needed */
+static int varlink_call_internal(sd_varlink *v, sd_json_variant *request) {
int r;
- assert_return(v, -EINVAL);
- assert_return(method, -EINVAL);
+ assert(v);
+ assert(request);
if (v->state == VARLINK_DISCONNECTED)
return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected.");
@@ -2281,14 +2291,7 @@ _public_ int sd_varlink_call_full(
* that we can assign a new reply shortly. */
varlink_clear_current(v);
- r = sd_json_buildo(
- &m,
- SD_JSON_BUILD_PAIR_STRING("method", method),
- JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters));
- if (r < 0)
- return varlink_log_errno(v, r, "Failed to build json message: %m");
-
- r = varlink_enqueue_json(v, m);
+ r = varlink_enqueue_json(v, request);
if (r < 0)
return varlink_log_errno(v, r, "Failed to enqueue json message: %m");
@@ -2310,29 +2313,9 @@ _public_ int sd_varlink_call_full(
switch (v->state) {
- case VARLINK_CALLED: {
+ case VARLINK_CALLED:
assert(v->current);
-
- varlink_set_state(v, VARLINK_IDLE_CLIENT);
- assert(v->n_pending == 1);
- v->n_pending--;
-
- sd_json_variant *e = sd_json_variant_by_key(v->current, "error"),
- *p = sd_json_variant_by_key(v->current, "parameters");
-
- /* If caller doesn't ask for the error string, then let's return an error code in case of failure */
- if (!ret_error_id && e)
- return sd_varlink_error_to_errno(sd_json_variant_string(e), p);
-
- if (ret_parameters)
- *ret_parameters = p;
- if (ret_error_id)
- *ret_error_id = e ? sd_json_variant_string(e) : NULL;
- if (ret_flags)
- *ret_flags = v->current_reply_flags;
-
- return 1;
- }
+ return 0;
case VARLINK_PENDING_DISCONNECT:
case VARLINK_DISCONNECTED:
@@ -2346,6 +2329,52 @@ _public_ int sd_varlink_call_full(
}
}
+_public_ int sd_varlink_call_full(
+ sd_varlink *v,
+ const char *method,
+ sd_json_variant *parameters,
+ sd_json_variant **ret_parameters,
+ const char **ret_error_id,
+ sd_varlink_reply_flags_t *ret_flags) {
+
+ _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
+ int r;
+
+ assert_return(v, -EINVAL);
+ assert_return(method, -EINVAL);
+
+ r = sd_json_buildo(
+ &m,
+ SD_JSON_BUILD_PAIR_STRING("method", method),
+ JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters));
+ if (r < 0)
+ return varlink_log_errno(v, r, "Failed to build json message: %m");
+
+ r = varlink_call_internal(v, m);
+ if (r < 0)
+ return r;
+
+ varlink_set_state(v, VARLINK_IDLE_CLIENT);
+ assert(v->n_pending == 1);
+ v->n_pending--;
+
+ sd_json_variant *e = sd_json_variant_by_key(v->current, "error"),
+ *p = sd_json_variant_by_key(v->current, "parameters");
+
+ /* If caller doesn't ask for the error string, then let's return an error code in case of failure */
+ if (!ret_error_id && e)
+ return sd_varlink_error_to_errno(sd_json_variant_string(e), p);
+
+ if (ret_parameters)
+ *ret_parameters = p;
+ if (ret_error_id)
+ *ret_error_id = e ? sd_json_variant_string(e) : NULL;
+ if (ret_flags)
+ *ret_flags = v->current_reply_flags;
+
+ return 1;
+}
+
_public_ int sd_varlink_call(
sd_varlink *v,
const char *method,
@@ -2356,6 +2385,116 @@ _public_ int sd_varlink_call(
return sd_varlink_call_full(v, method, parameters, ret_parameters, ret_error_id, NULL);
}
+_public_ int sd_varlink_call_and_upgrade(
+ sd_varlink *v,
+ const char *method,
+ sd_json_variant *parameters,
+ sd_json_variant **ret_parameters,
+ const char **ret_error_id,
+ int *ret_input_fd,
+ int *ret_output_fd) {
+
+ _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL;
+ int r;
+
+ assert_return(v, -EINVAL);
+ assert_return(method, -EINVAL);
+ assert_return(ret_input_fd || ret_output_fd, -EINVAL);
+
+ r = sd_json_buildo(
+ &m,
+ SD_JSON_BUILD_PAIR_STRING("method", method),
+ JSON_BUILD_PAIR_VARIANT_NON_EMPTY("parameters", parameters),
+ SD_JSON_BUILD_PAIR_BOOLEAN("upgrade", true));
+ if (r < 0)
+ return varlink_log_errno(v, r, "Failed to build json message: %m");
+
+ v->protocol_upgrade = true;
+ r = varlink_call_internal(v, m);
+ if (r < 0) {
+ v->protocol_upgrade = false;
+ return r;
+ }
+
+ /* ensure we did not consume any data from the upgraded protocol */
+ assert(v->input_buffer_size == 0);
+
+ sd_json_variant *e = sd_json_variant_by_key(v->current, "error"),
+ *p = sd_json_variant_by_key(v->current, "parameters");
+
+ /* don't steal the fd on server error */
+ if (e) {
+ if (ret_error_id) {
+ *ret_error_id = sd_json_variant_string(e);
+ if (ret_parameters)
+ *ret_parameters = p;
+ r = 0;
+ } else
+ r = sd_varlink_error_to_errno(sd_json_variant_string(e), p);
+
+ varlink_set_state(v, VARLINK_IDLE_CLIENT);
+ goto finish;
+ }
+
+ /* Pass the connection fds to the caller, it owns them now. Reset to blocking mode
+ * since callers of the upgraded protocol will generally expect normal blocking
+ * semantics. */
+ r = fd_nonblock(v->input_fd, false);
+ if (r < 0) {
+ varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m");
+ goto disconnect;
+ }
+ if (v->input_fd != v->output_fd) {
+ r = fd_nonblock(v->output_fd, false);
+ if (r < 0) {
+ varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m");
+ goto disconnect;
+ }
+ }
+
+ /* Handle the case where the caller is not interested in one of the fds. We need
+ * to consider the case when (input_fd == output_fd) just clear the alias
+ * rather than closing it, since the other branch may hand it out. */
+ bool same_fd = v->input_fd == v->output_fd;
+
+ if (ret_input_fd)
+ *ret_input_fd = TAKE_FD(v->input_fd);
+ else if (!same_fd) {
+ (void) shutdown(v->input_fd, SHUT_RD);
+ v->input_fd = safe_close(v->input_fd);
+ } else
+ v->input_fd = -EBADF;
+
+ if (ret_output_fd)
+ *ret_output_fd = TAKE_FD(v->output_fd);
+ else if (!same_fd) {
+ (void) shutdown(v->output_fd, SHUT_WR);
+ v->output_fd = safe_close(v->output_fd);
+ } else
+ v->output_fd = -EBADF;
+
+ varlink_set_state(v, VARLINK_DISCONNECTED);
+ assert(v->n_pending == 1);
+ v->n_pending--;
+
+ if (ret_parameters)
+ *ret_parameters = p;
+ if (ret_error_id)
+ *ret_error_id = NULL;
+
+ return 1;
+
+disconnect:
+ /* If we fail after the server already accepted the upgrade, nothing can be done but disconnect.
+ * The other side is speaking raw protocol while we expect JSON. */
+ varlink_set_state(v, VARLINK_DISCONNECTED);
+finish:
+ v->protocol_upgrade = false;
+ assert(v->n_pending == 1);
+ v->n_pending--;
+ return r;
+}
+
_public_ int sd_varlink_callb_ap(
sd_varlink *v,
const char *method,
diff --git a/src/libsystemd/sd-varlink/varlink-internal.h b/src/libsystemd/sd-varlink/varlink-internal.h
index 39b15e12a1..3dfe86d448 100644
--- a/src/libsystemd/sd-varlink/varlink-internal.h
+++ b/src/libsystemd/sd-varlink/varlink-internal.h
@@ -170,6 +170,8 @@ typedef struct sd_varlink {
bool prefer_write:1;
bool got_pollhup:1;
+ bool protocol_upgrade:1; /* Whether a protocol_upgrade was requested */
+
bool output_buffer_sensitive:1; /* whether to erase the output buffer after writing it to the socket */
bool input_sensitive:1; /* Whether incoming messages might be sensitive */
diff --git a/src/shared/socket-forward.c b/src/shared/socket-forward.c
index 2601b25e6d..10c0a9be67 100644
--- a/src/shared/socket-forward.c
+++ b/src/shared/socket-forward.c
@@ -1,6 +1,7 @@
/* SPDX-License-Identifier: LGPL-2.1-or-later */
#include
+#include
#include
#include "sd-event.h"
@@ -13,41 +14,42 @@
#define SOCKET_FORWARD_BUFFER_SIZE (256 * 1024)
-struct SocketForward {
+/* Unidirectional forwarder: splices data from read_fd to write_fd via a kernel pipe buffer.
+ * Each direction of a full-duplex SocketForward is handled by one of these. */
+typedef struct SimplexForward {
sd_event *event;
- int server_fd, client_fd;
+ int read_fd, write_fd;
- int server_to_client_buffer[2]; /* a pipe */
- int client_to_server_buffer[2]; /* a pipe */
+ int buffer[2]; /* a pipe */
- size_t server_to_client_buffer_full, client_to_server_buffer_full;
- size_t server_to_client_buffer_size, client_to_server_buffer_size;
+ size_t buffer_full, buffer_size;
- sd_event_source *server_event_source, *client_event_source;
+ sd_event_source *read_event_source, *write_event_source;
- socket_forward_done_t on_done;
+ int (*on_done)(struct SimplexForward *fwd, int error, void *userdata);
void *userdata;
-};
+} SimplexForward;
-SocketForward* socket_forward_free(SocketForward *sf) {
- if (!sf)
+static SimplexForward* simplex_forward_free(SimplexForward *fwd) {
+ if (!fwd)
return NULL;
- sd_event_source_unref(sf->server_event_source);
- sd_event_source_unref(sf->client_event_source);
+ sd_event_source_unref(fwd->read_event_source);
+ sd_event_source_unref(fwd->write_event_source);
- safe_close(sf->server_fd);
- safe_close(sf->client_fd);
+ safe_close(fwd->read_fd);
+ safe_close(fwd->write_fd);
- safe_close_pair(sf->server_to_client_buffer);
- safe_close_pair(sf->client_to_server_buffer);
+ safe_close_pair(fwd->buffer);
- sd_event_unref(sf->event);
+ sd_event_unref(fwd->event);
- return mfree(sf);
+ return mfree(fwd);
}
+DEFINE_TRIVIAL_CLEANUP_FUNC(SimplexForward*, simplex_forward_free);
+
static int socket_forward_create_pipes(int buffer[static 2], size_t *ret_size) {
int r;
@@ -73,7 +75,7 @@ static int socket_forward_create_pipes(int buffer[static 2], size_t *ret_size) {
return 0;
}
-static int socket_forward_shovel(
+static int simplex_forward_shovel(
int *from, int buffer[2], int *to,
size_t *full, size_t *sz,
sd_event_source **from_source, sd_event_source **to_source) {
@@ -123,85 +125,251 @@ static int socket_forward_shovel(
return 0;
}
-static int socket_forward_enable_event_sources(SocketForward *sf);
+static int simplex_forward_enable_event_sources(SimplexForward *fwd);
-static int socket_forward_traffic_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
- SocketForward *sf = ASSERT_PTR(userdata);
+static int simplex_forward_traffic(SimplexForward *fwd) {
int r;
- assert(s);
- assert(fd >= 0);
-
- r = socket_forward_shovel(
- &sf->server_fd, sf->server_to_client_buffer, &sf->client_fd,
- &sf->server_to_client_buffer_full, &sf->server_to_client_buffer_size,
- &sf->server_event_source, &sf->client_event_source);
+ r = simplex_forward_shovel(
+ &fwd->read_fd, fwd->buffer, &fwd->write_fd,
+ &fwd->buffer_full, &fwd->buffer_size,
+ &fwd->read_event_source, &fwd->write_event_source);
if (r < 0)
goto quit;
- r = socket_forward_shovel(
- &sf->client_fd, sf->client_to_server_buffer, &sf->server_fd,
- &sf->client_to_server_buffer_full, &sf->client_to_server_buffer_size,
- &sf->client_event_source, &sf->server_event_source);
- if (r < 0)
+ /* Read side closed and all buffered data written? */
+ if (fwd->read_fd < 0 && fwd->buffer_full <= 0)
goto quit;
- /* EOF on both sides? */
- if (sf->server_fd < 0 && sf->client_fd < 0)
+ /* Write side closed? */
+ if (fwd->write_fd < 0)
goto quit;
- /* Server closed, and all data written to client? */
- if (sf->server_fd < 0 && sf->server_to_client_buffer_full <= 0)
- goto quit;
-
- /* Client closed, and all data written to server? */
- if (sf->client_fd < 0 && sf->client_to_server_buffer_full <= 0)
- goto quit;
-
- r = socket_forward_enable_event_sources(sf);
+ r = simplex_forward_enable_event_sources(fwd);
if (r < 0)
goto quit;
return 1;
quit:
- return sf->on_done(sf, r, sf->userdata);
+ fwd->read_event_source = sd_event_source_disable_unref(fwd->read_event_source);
+ fwd->write_event_source = sd_event_source_disable_unref(fwd->write_event_source);
+ return fwd->on_done(fwd, r, fwd->userdata);
}
-static int socket_forward_enable_event_sources(SocketForward *sf) {
- uint32_t a = 0, b = 0;
+static int simplex_forward_io_cb(sd_event_source *s, int fd, uint32_t revents, void *userdata) {
+ SimplexForward *fwd = ASSERT_PTR(userdata);
+
+ return simplex_forward_traffic(fwd);
+}
+
+static int simplex_forward_defer_cb(sd_event_source *s, void *userdata) {
+ SimplexForward *fwd = ASSERT_PTR(userdata);
+
+ return simplex_forward_traffic(fwd);
+}
+
+static int simplex_forward_enable_event_sources(SimplexForward *fwd) {
+ bool can_read, can_write;
int r;
- assert(sf);
+ assert(fwd);
- if (sf->server_to_client_buffer_full > 0)
- b |= EPOLLOUT;
- if (sf->server_to_client_buffer_full < sf->server_to_client_buffer_size)
- a |= EPOLLIN;
+ can_read = fwd->buffer_full < fwd->buffer_size;
+ can_write = fwd->buffer_full > 0;
- if (sf->client_to_server_buffer_full > 0)
- a |= EPOLLOUT;
- if (sf->client_to_server_buffer_full < sf->client_to_server_buffer_size)
- b |= EPOLLIN;
+ /* Event sources may have been unref'd by the shovel on EOF/disconnect */
+ if (fwd->read_event_source) {
+ r = sd_event_source_set_enabled(fwd->read_event_source, can_read ? SD_EVENT_ONESHOT : SD_EVENT_OFF);
+ if (r < 0)
+ return log_debug_errno(r, "Failed to update read event source: %m");
+ }
- if (sf->server_event_source)
- r = sd_event_source_set_io_events(sf->server_event_source, a);
- else if (sf->server_fd >= 0)
- r = sd_event_add_io(sf->event, &sf->server_event_source, sf->server_fd, a, socket_forward_traffic_cb, sf);
- else
- r = 0;
+ if (fwd->write_event_source) {
+ r = sd_event_source_set_enabled(fwd->write_event_source, can_write ? SD_EVENT_ONESHOT : SD_EVENT_OFF);
+ if (r < 0)
+ return log_debug_errno(r, "Failed to update write event source: %m");
+ }
+
+ return 0;
+}
+
+static int simplex_forward_create_event_source(
+ SimplexForward *fwd,
+ sd_event_source **ret,
+ int fd,
+ uint32_t events) {
+
+ int r;
+
+ r = sd_event_add_io(fwd->event, ret, fd, events, simplex_forward_io_cb, fwd);
+ if (r == -EPERM)
+ /* fd is not pollable (e.g. regular file). Fall back to a defer event source
+ * which fires on each event loop iteration. This works because regular
+ * file are always ready for I/O so we don't need to poll. */
+ r = sd_event_add_defer(fwd->event, ret, simplex_forward_defer_cb, fwd);
+
+ return r;
+}
+
+static int simplex_forward_new(
+ sd_event *event,
+ int read_fd,
+ int write_fd,
+ int (*on_done)(SimplexForward *fwd, int error, void *userdata),
+ void *userdata,
+ SimplexForward **ret) {
+
+ _cleanup_(simplex_forward_freep) SimplexForward *fwd = NULL;
+ int r;
+
+ assert(event);
+ assert(read_fd >= 0);
+ assert(write_fd >= 0);
+ assert(read_fd != write_fd);
+ assert(on_done);
+ assert(ret);
+
+ fwd = new(SimplexForward, 1);
+ if (!fwd) {
+ safe_close(read_fd);
+ safe_close(write_fd);
+ return log_oom_debug();
+ }
+
+ *fwd = (SimplexForward) {
+ .event = sd_event_ref(event),
+ .read_fd = read_fd,
+ .write_fd = write_fd,
+ .buffer = EBADF_PAIR,
+ .on_done = on_done,
+ .userdata = userdata,
+ };
+
+ r = socket_forward_create_pipes(fwd->buffer, &fwd->buffer_size);
if (r < 0)
- return log_debug_errno(r, "Failed to set up server event source: %m");
+ return r;
- if (sf->client_event_source)
- r = sd_event_source_set_io_events(sf->client_event_source, b);
- else if (sf->client_fd >= 0)
- r = sd_event_add_io(sf->event, &sf->client_event_source, sf->client_fd, b, socket_forward_traffic_cb, sf);
- else
- r = 0;
+ r = simplex_forward_create_event_source(fwd, &fwd->read_event_source, fwd->read_fd, EPOLLIN);
if (r < 0)
- return log_debug_errno(r, "Failed to set up client event source: %m");
+ return r;
+ r = simplex_forward_create_event_source(fwd, &fwd->write_event_source, fwd->write_fd, EPOLLOUT);
+ if (r < 0)
+ return r;
+
+ r = simplex_forward_enable_event_sources(fwd);
+ if (r < 0)
+ return r;
+
+ *ret = TAKE_PTR(fwd);
+ return 0;
+}
+
+/* Full-duplex forwarder from two SimplexForward instances */
+struct SocketForward {
+ SimplexForward *server_to_client;
+ SimplexForward *client_to_server;
+
+ socket_forward_done_t on_done;
+ void *userdata;
+
+ int first_error;
+ unsigned directions_done;
+};
+
+SocketForward* socket_forward_free(SocketForward *sf) {
+ if (!sf)
+ return NULL;
+
+ simplex_forward_free(sf->server_to_client);
+ simplex_forward_free(sf->client_to_server);
+
+ return mfree(sf);
+}
+
+static int socket_forward_direction_done(SimplexForward *fwd, int error, void *userdata) {
+ SocketForward *sf = ASSERT_PTR(userdata);
+
+ /* Half-close the write side so the remote end sees EOF. For sockets,
+ * shutdown(SHUT_WR) sends FIN while keeping the fd open for the read side
+ * (which belongs to the other direction's dup'd fd). For pipes/FIFOs,
+ * shutdown() fails with ENOTSOCK - close the fd instead, which is the
+ * only way to signal EOF on a pipe. */
+ if (fwd->write_fd >= 0 && shutdown(fwd->write_fd, SHUT_WR) < 0) {
+ if (errno == ENOTSOCK)
+ fwd->write_fd = safe_close(fwd->write_fd);
+ else
+ log_debug_errno(errno, "Failed to shutdown write side of fd %d: %m, ignoring",
+ fwd->write_fd);
+ }
+
+ if (error < 0 && sf->first_error >= 0)
+ sf->first_error = error;
+
+ sf->directions_done++;
+
+ if (sf->directions_done >= 2)
+ return sf->on_done(sf, sf->first_error, sf->userdata);
+
+ return 0;
+}
+
+int socket_forward_new_pair(
+ sd_event *event,
+ int server_read_fd,
+ int server_write_fd,
+ int client_read_fd,
+ int client_write_fd,
+ socket_forward_done_t on_done,
+ void *userdata,
+ SocketForward **ret) {
+
+ _cleanup_close_ int server_read_fd_close = server_read_fd,
+ server_write_fd_close = server_write_fd,
+ client_read_fd_close = client_read_fd,
+ client_write_fd_close = client_write_fd;
+ _cleanup_(socket_forward_freep) SocketForward *sf = NULL;
+ int r;
+
+ assert(event);
+ assert(server_read_fd >= 0);
+ assert(server_write_fd >= 0);
+ assert(client_read_fd >= 0);
+ assert(client_write_fd >= 0);
+ assert(server_read_fd != server_write_fd);
+ assert(client_read_fd != client_write_fd);
+ assert(server_read_fd != client_read_fd);
+ assert(server_read_fd != client_write_fd);
+ assert(server_write_fd != client_read_fd);
+ assert(server_write_fd != client_write_fd);
+ assert(on_done);
+ assert(ret);
+
+ sf = new(SocketForward, 1);
+ if (!sf)
+ return log_oom_debug();
+
+ *sf = (SocketForward) {
+ .on_done = on_done,
+ .userdata = userdata,
+ };
+
+ r = simplex_forward_new(event,
+ TAKE_FD(server_read_fd_close), TAKE_FD(client_write_fd_close),
+ socket_forward_direction_done, sf,
+ &sf->server_to_client);
+ if (r < 0)
+ return r;
+
+ r = simplex_forward_new(event,
+ TAKE_FD(client_read_fd_close), TAKE_FD(server_write_fd_close),
+ socket_forward_direction_done, sf,
+ &sf->client_to_server);
+ if (r < 0)
+ return r;
+
+ *ret = TAKE_PTR(sf);
return 0;
}
@@ -213,8 +381,8 @@ int socket_forward_new(
void *userdata,
SocketForward **ret) {
- _cleanup_(socket_forward_freep) SocketForward *sf = NULL;
- int r;
+ _cleanup_close_ int server_fd_close = server_fd, client_fd_close = client_fd,
+ server_write_fd = -EBADF, client_write_fd = -EBADF;
assert(event);
assert(server_fd >= 0);
@@ -222,35 +390,17 @@ int socket_forward_new(
assert(on_done);
assert(ret);
- sf = new(SocketForward, 1);
- if (!sf) {
- safe_close(server_fd);
- safe_close(client_fd);
- return log_oom_debug();
- }
+ server_write_fd = fcntl(server_fd, F_DUPFD_CLOEXEC, 3);
+ if (server_write_fd < 0)
+ return -errno;
- *sf = (SocketForward) {
- .event = sd_event_ref(event),
- .server_fd = server_fd,
- .client_fd = client_fd,
- .server_to_client_buffer = EBADF_PAIR,
- .client_to_server_buffer = EBADF_PAIR,
- .on_done = on_done,
- .userdata = userdata,
- };
+ client_write_fd = fcntl(client_fd, F_DUPFD_CLOEXEC, 3);
+ if (client_write_fd < 0)
+ return -errno;
- r = socket_forward_create_pipes(sf->server_to_client_buffer, &sf->server_to_client_buffer_size);
- if (r < 0)
- return r;
-
- r = socket_forward_create_pipes(sf->client_to_server_buffer, &sf->client_to_server_buffer_size);
- if (r < 0)
- return r;
-
- r = socket_forward_enable_event_sources(sf);
- if (r < 0)
- return r;
-
- *ret = TAKE_PTR(sf);
- return 0;
+ return socket_forward_new_pair(
+ event,
+ TAKE_FD(server_fd_close), TAKE_FD(server_write_fd),
+ TAKE_FD(client_fd_close), TAKE_FD(client_write_fd),
+ on_done, userdata, ret);
}
diff --git a/src/shared/socket-forward.h b/src/shared/socket-forward.h
index a2d34da38c..a5b81b4a44 100644
--- a/src/shared/socket-forward.h
+++ b/src/shared/socket-forward.h
@@ -3,20 +3,22 @@
#include "shared-forward.h"
-/* Bidirectional socket forwarder using splice().
+/* Bidirectional forwarder using splice().
*
- * Forwards data between two bidirectional sockets ("server" and "client") via kernel pipe buffers,
- * avoiding userspace copies.
+ * Forwards data between two sides ("server" and "client") via kernel pipe buffers,
+ * avoiding userspace copies. Internally uses two independent half-duplex forwarders,
+ * one per direction. All four fds must be distinct - use dup()/fcntl(fd, F_DUPFD_CLOEXEC, 3) for bidirectional sockets.
*
* When forwarding completes (both directions reach EOF or error), the completion callback is invoked.
*
- * The SocketForward takes ownership of both fds - they are closed when the SocketForward is freed
+ * The SocketForward takes ownership of all fds - they are closed when the SocketForward is freed
* (or earlier, during normal forwarding when EOF/disconnect is detected). */
typedef struct SocketForward SocketForward;
typedef int (*socket_forward_done_t)(SocketForward *sf, int error, void *userdata);
+/* Create a forwarder between two bidirectional sockets. */
int socket_forward_new(
sd_event *event,
int server_fd,
@@ -25,5 +27,17 @@ int socket_forward_new(
void *userdata,
SocketForward **ret);
+/* Create a forwarder between two fd pairs (e.g. stdin/stdout on one side, socket on the other).
+ * All four fds must be distinct - use dup()/fcntl(fd, F_DUPFD_CLOEXEC, 3) for bidirectional sockets. */
+int socket_forward_new_pair(
+ sd_event *event,
+ int server_read_fd,
+ int server_write_fd,
+ int client_read_fd,
+ int client_write_fd,
+ socket_forward_done_t on_done,
+ void *userdata,
+ SocketForward **ret);
+
SocketForward* socket_forward_free(SocketForward *sf);
DEFINE_TRIVIAL_CLEANUP_FUNC(SocketForward*, socket_forward_free);
diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h
index 7cda9e7e56..9d6e939d64 100644
--- a/src/systemd/sd-varlink.h
+++ b/src/systemd/sd-varlink.h
@@ -55,8 +55,9 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_reply_flags_t) {
} sd_varlink_reply_flags_t;
__extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_method_flags_t) {
- SD_VARLINK_METHOD_ONEWAY = 1 << 0,
- SD_VARLINK_METHOD_MORE = 1 << 1,
+ SD_VARLINK_METHOD_ONEWAY = 1 << 0,
+ SD_VARLINK_METHOD_MORE = 1 << 1,
+ SD_VARLINK_METHOD_UPGRADE = 1 << 2,
_SD_ENUM_FORCE_S64(SD_VARLINK_METHOD)
} sd_varlink_method_flags_t;
@@ -135,6 +136,13 @@ int sd_varlink_callb(sd_varlink *v, const char *method, sd_json_variant **ret_pa
#define sd_varlink_callbo(v, method, ret_parameters, ret_error_id, ...) \
sd_varlink_callb((v), (method), (ret_parameters), (ret_error_id), SD_JSON_BUILD_OBJECT(__VA_ARGS__))
+/* Send method call with upgrade, wait for reply, then steal the connection fds for raw I/O.
+ * For bidirectional sockets ret_input_fd and ret_output_fd will be the same fd.
+ * ret_parameters and ret_error_id are borrowed references valid only until v is closed or unreffed.
+ * Returns > 0 if the connection was upgraded, 0 if a Varlink error occurred (and ret_error_id was set),
+ * or < 0 on local failure. */
+int sd_varlink_call_and_upgrade(sd_varlink *v, const char *method, sd_json_variant *parameters, sd_json_variant **ret_parameters, const char **ret_error_id, int *ret_input_fd, int *ret_output_fd);
+
/* Send method call and begin collecting all 'more' replies into an array, finishing when a final reply is sent */
int sd_varlink_collect_full(sd_varlink *v, const char *method, sd_json_variant *parameters, sd_json_variant **ret_parameters, const char **ret_error_id, sd_varlink_reply_flags_t *ret_flags);
int sd_varlink_collect(sd_varlink *v, const char *method, sd_json_variant *parameters, sd_json_variant **ret_parameters, const char **ret_error_id);
diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c
index 775a481ae8..4495d090df 100644
--- a/src/varlinkctl/varlinkctl.c
+++ b/src/varlinkctl/varlinkctl.c
@@ -12,6 +12,7 @@
#include "bus-util.h"
#include "chase.h"
#include "env-util.h"
+#include "escape.h"
#include "fd-util.h"
#include "fileio.h"
#include "format-table.h"
@@ -30,6 +31,7 @@
#include "process-util.h"
#include "recurse-dir.h"
#include "runtime-scope.h"
+#include "socket-forward.h"
#include "string-util.h"
#include "strv.h"
#include "terminal-util.h"
@@ -52,6 +54,7 @@ static bool arg_quiet = false;
static char **arg_graceful = NULL;
static usec_t arg_timeout = 0;
static bool arg_exec = false;
+static bool arg_upgrade = false;
static PushFds arg_push_fds = {};
static bool arg_ask_password = true;
static bool arg_legend = true;
@@ -111,6 +114,8 @@ static int help(void) {
" --graceful=ERROR Treat specified Varlink error as success\n"
" --timeout=SECS Maximum time to wait for method call completion\n"
" -E Short for --more --timeout=infinity\n"
+ " --upgrade Request protocol upgrade (connection becomes raw\n"
+ " bidirectional pipe on stdin/stdout after reply)\n"
" --push-fd=FD Pass the specified fd along with method call\n"
"\nSee the %2$s for details.\n",
program_invocation_short_name,
@@ -139,6 +144,7 @@ static int parse_argv(int argc, char *argv[]) {
ARG_GRACEFUL,
ARG_TIMEOUT,
ARG_EXEC,
+ ARG_UPGRADE,
ARG_PUSH_FD,
ARG_NO_ASK_PASSWORD,
ARG_USER,
@@ -157,6 +163,7 @@ static int parse_argv(int argc, char *argv[]) {
{ "graceful", required_argument, NULL, ARG_GRACEFUL },
{ "timeout", required_argument, NULL, ARG_TIMEOUT },
{ "exec", no_argument, NULL, ARG_EXEC },
+ { "upgrade", no_argument, NULL, ARG_UPGRADE },
{ "push-fd", required_argument, NULL, ARG_PUSH_FD },
{ "no-ask-password", no_argument, NULL, ARG_NO_ASK_PASSWORD },
{ "user", no_argument, NULL, ARG_USER },
@@ -245,6 +252,10 @@ static int parse_argv(int argc, char *argv[]) {
arg_exec = true;
break;
+ case ARG_UPGRADE:
+ arg_upgrade = true;
+ break;
+
case ARG_PUSH_FD: {
if (!GREEDY_REALLOC(arg_push_fds.fds, arg_push_fds.n_fds + 1))
return log_oom();
@@ -634,11 +645,179 @@ static int reply_callback(
return r;
}
+static int upgrade_forward_done(SocketForward *sf, int error, void *userdata) {
+ sd_event *event = ASSERT_PTR(userdata);
+
+ return sd_event_exit(event, error < 0 ? error : 0);
+}
+
+_noreturn_ static void exec_with_listen_fds(char **exec_cmdline, int *fds, size_t n_fds) {
+ int r;
+
+ _cleanup_free_ char *j = quote_command_line(exec_cmdline, SHELL_ESCAPE_EMPTY);
+ if (!j) {
+ log_oom();
+ _exit(EXIT_FAILURE);
+ }
+
+ log_close();
+ log_set_open_when_needed(true);
+
+ r = close_all_fds(fds, n_fds);
+ if (r < 0) {
+ log_error_errno(r, "Failed to close all remaining file descriptors: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ r = pack_fds(fds, n_fds);
+ if (r < 0) {
+ log_error_errno(r, "Failed to rearrange file descriptors: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ r = fd_cloexec_many(fds, n_fds, false);
+ if (r < 0) {
+ log_error_errno(r, "Failed to disable O_CLOEXEC for file descriptors: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ if (n_fds > 0) {
+ r = setenvf("LISTEN_FDS", /* overwrite= */ true, "%zu", n_fds);
+ if (r < 0) {
+ log_error_errno(r, "Failed to set $LISTEN_FDS environment variable: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ r = setenvf("LISTEN_PID", /* overwrite= */ true, PID_FMT, getpid_cached());
+ if (r < 0) {
+ log_error_errno(r, "Failed to set $LISTEN_PID environment variable: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ uint64_t pidfdid;
+ if (pidfd_get_inode_id_self_cached(&pidfdid) >= 0) {
+ r = setenvf("LISTEN_PIDFDID", /* overwrite= */ true, "%" PRIu64, pidfdid);
+ if (r < 0) {
+ log_error_errno(r, "Failed to set $LISTEN_PIDFDID environment variable: %m");
+ _exit(EXIT_FAILURE);
+ }
+ }
+ } else {
+ (void) unsetenv("LISTEN_FDS");
+ (void) unsetenv("LISTEN_PID");
+ (void) unsetenv("LISTEN_PIDFDID");
+ }
+ (void) unsetenv("LISTEN_FDNAMES");
+
+ log_debug("Executing: %s", j);
+
+ execvp(exec_cmdline[0], exec_cmdline);
+ log_error_errno(errno, "Failed to execute '%s': %m", j);
+ _exit(EXIT_FAILURE);
+}
+
+static int varlink_call_and_upgrade(const char *url, const char *method, sd_json_variant *parameters, char **exec_cmdline) {
+ _cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;
+ _cleanup_(sd_event_unrefp) sd_event *event = NULL;
+ _cleanup_(socket_forward_freep) SocketForward *sf = NULL;
+ _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF;
+ const char *error_id = NULL;
+ int r;
+
+ r = varlink_connect_auto(&vl, url);
+ if (r < 0)
+ return r;
+
+ r = sd_varlink_call_and_upgrade(
+ vl,
+ method,
+ parameters,
+ /* ret_parameters= */ NULL,
+ &error_id,
+ &input_fd,
+ &output_fd);
+ if (r < 0)
+ return log_error_errno(r, "Failed to upgrade connection via %s(): %m", method);
+ if (!isempty(error_id))
+ return log_error_errno(SYNTHETIC_ERRNO(EBADE), "Upgrade via %s() failed with error: %s", method, error_id);
+
+ /* For bidirectional sockets input_fd == output_fd. Dup immediately so that _cleanup_close_
+ * on both variables can never double-close the same fd. Note that on fcntl() failure
+ * output_fd is overwritten with -1, so only input_fd holds the real fd at cleanup time. */
+ if (input_fd == output_fd) {
+ output_fd = fcntl(input_fd, F_DUPFD_CLOEXEC, 3);
+ if (output_fd < 0)
+ return log_error_errno(errno, "Failed to dup upgraded connection fd: %m");
+ }
+
+ if (!strv_isempty(exec_cmdline)) {
+ /* --exec mode: place the upgraded connection on stdin/stdout so that the child
+ * process can just read/write naturally. */
+ (void) sd_notify(/* unset_environment= */ false, "READY=1");
+
+ log_close();
+ log_set_open_when_needed(true);
+
+ int keep_fds[] = { TAKE_FD(input_fd), TAKE_FD(output_fd) };
+
+ r = close_all_fds(keep_fds, 2);
+ if (r < 0) {
+ log_error_errno(r, "Failed to close remaining file descriptors: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ r = move_fd(keep_fds[0], STDIN_FILENO, /* cloexec= */ false);
+ if (r < 0) {
+ log_error_errno(r, "Failed to move upgraded connection to stdin: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ r = move_fd(keep_fds[1], STDOUT_FILENO, /* cloexec= */ false);
+ if (r < 0) {
+ log_error_errno(r, "Failed to move upgraded connection to stdout: %m");
+ _exit(EXIT_FAILURE);
+ }
+
+ /* We exec and never return here */
+ exec_with_listen_fds(exec_cmdline, /* fds= */ NULL, /* n_fds= */ 0);
+ }
+
+ /* No --exec: bidirectional proxy between stdin/stdout and the upgraded socket */
+
+ /* Use a clean event loop just for the forwarding to avoid any interference. */
+ r = sd_event_new(&event);
+ if (r < 0)
+ return log_error_errno(r, "Failed to allocate event loop: %m");
+
+ _cleanup_close_ int stdin_fd = fcntl(STDIN_FILENO, F_DUPFD_CLOEXEC, 3);
+ if (stdin_fd < 0)
+ return log_error_errno(errno, "Failed to dup stdin: %m");
+
+ _cleanup_close_ int stdout_fd = fcntl(STDOUT_FILENO, F_DUPFD_CLOEXEC, 3);
+ if (stdout_fd < 0)
+ return log_error_errno(errno, "Failed to dup stdout: %m");
+
+ r = socket_forward_new_pair(
+ event,
+ TAKE_FD(stdin_fd), TAKE_FD(stdout_fd),
+ TAKE_FD(input_fd), TAKE_FD(output_fd),
+ upgrade_forward_done, event,
+ &sf);
+ if (r < 0)
+ return log_error_errno(r, "Failed to set up socket forwarding for varlink: %m");
+
+ r = sd_event_loop(event);
+ if (r < 0)
+ return log_error_errno(r, "Failed to run socket forward for varlink: %m");
+
+ return 0;
+}
+
static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) {
_cleanup_(sd_json_variant_unrefp) sd_json_variant *jp = NULL;
_cleanup_(sd_varlink_unrefp) sd_varlink *vl = NULL;
const char *url, *method, *parameter, *source;
- char **cmdline;
+ char **exec_cmdline;
int r;
assert(argc >= 3);
@@ -651,12 +830,31 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) {
if (arg_exec && (arg_collect || (arg_method_flags & (SD_VARLINK_METHOD_ONEWAY|SD_VARLINK_METHOD_MORE))) != 0)
return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "--exec and --collect/--more/--oneway may not be combined.");
+ if (arg_upgrade) {
+ if (arg_collect || arg_method_flags != 0 || arg_push_fds.n_fds > 0 || !strv_isempty(arg_graceful))
+ return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "--upgrade may not be combined with --collect/--more/--oneway/--push-fd=/--graceful.");
+ }
+
(void) polkit_agent_open_if_enabled(BUS_TRANSPORT_LOCAL, arg_ask_password);
url = argv[1];
method = argv[2];
parameter = argc > 3 && !streq(argv[3], "-") ? argv[3] : NULL;
- cmdline = strv_skip(argv, 4);
+ exec_cmdline = strv_skip(argv, 4);
+
+ if (!varlink_idl_qualified_symbol_name_is_valid(method))
+ return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Not a valid qualified method name: '%s' (Expected valid Varlink interface name, followed by a dot, followed by a valid Varlink symbol name.)", method);
+
+ if (arg_upgrade) {
+ /* For --upgrade, parse parameters from argv only (stdin is used for the upgraded connection) */
+ if (parameter) {
+ r = sd_json_parse(parameter, SD_JSON_PARSE_MUST_BE_OBJECT, &jp, /* reterr_line= */ NULL, /* reterr_column= */ NULL);
+ if (r < 0)
+ return log_error_errno(r, "Failed to parse parameters: %m");
+ }
+
+ return varlink_call_and_upgrade(url, method, jp, exec_cmdline);
+ }
/* No JSON mode explicitly configured? Then default to the same as -j (except if --exec is used, in
* which case generate shortest possible JSON since we are going to pass it to a program rather than
@@ -674,9 +872,6 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) {
* leave incomplete lines hanging around. */
arg_json_format_flags |= SD_JSON_FORMAT_NEWLINE;
- if (!varlink_idl_qualified_symbol_name_is_valid(method))
- return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Not a valid qualified method name: '%s' (Expected valid Varlink interface name, followed by a dot, followed by a valid Varlink symbol name.)", method);
-
unsigned line = 0, column = 0;
if (parameter) {
source = "";
@@ -842,10 +1037,6 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) {
if (mfd < 0)
return log_error_errno(mfd, "Failed to allocate memfd for reply: %m");
- _cleanup_free_ char *j = strv_join(cmdline, " ");
- if (!j)
- return log_oom();
-
int *fd_array = NULL, n = 0;
size_t m = 0;
CLEANUP_ARRAY(fd_array, m, close_many_and_free);
@@ -872,66 +1063,13 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) {
* lives in our process their fds. Hence we will now no longer bubble up any
* errors. */
- log_close();
- log_set_open_when_needed(true);
-
r = move_fd(mfd, STDIN_FILENO, /* cloexec= */ false);
if (r < 0) {
log_error_errno(r, "Failed to move reply to STDIN_FILENO: %m");
_exit(EXIT_FAILURE);
}
- r = close_all_fds(fd_array, m);
- if (r < 0) {
- log_error_errno(r, "Failed to close all remaining file descriptors: %m");
- _exit(EXIT_FAILURE);
- }
-
- r = pack_fds(fd_array, m);
- if (r < 0) {
- log_error_errno(r, "Failed to rearrange file descriptors: %m");
- _exit(EXIT_FAILURE);
- }
-
- r = fd_cloexec_many(fd_array, m, false);
- if (r < 0) {
- log_error_errno(r, "Failed to disable O_CLOEXEC for file descriptors: %m");
- _exit(EXIT_FAILURE);
- }
-
- if (m > 0) {
- r = setenvf("LISTEN_FDS", /* overwrite= */ true, "%zu", m);
- if (r < 0) {
- log_error_errno(r, "Failed to set $LISTEN_FDS environment variable: %m");
- _exit(EXIT_FAILURE);
- }
-
- r = setenvf("LISTEN_PID", /* overwrite= */ true, PID_FMT, getpid_cached());
- if (r < 0) {
- log_error_errno(r, "Failed to set $LISTEN_PID environment variable: %m");
- _exit(EXIT_FAILURE);
- }
-
- uint64_t pidfdid;
- if (pidfd_get_inode_id_self_cached(&pidfdid) >= 0) {
- r = setenvf("LISTEN_PIDFDID", /* overwrite= */ true, "%" PRIu64, pidfdid);
- if (r < 0) {
- log_error_errno(r, "Failed to set $LISTEN_PIDFDID environment variable: %m");
- _exit(EXIT_FAILURE);
- }
- }
- } else {
- (void) unsetenv("LISTEN_FDS");
- (void) unsetenv("LISTEN_PID");
- (void) unsetenv("LISTEN_PIDFDID");
- }
- (void) unsetenv("LISTEN_FDNAMES");
-
- log_debug("Executing: %s", j);
-
- execvp(cmdline[0], cmdline);
- log_error_errno(errno, "Failed to execute '%s': %m", j);
- _exit(EXIT_FAILURE);
+ exec_with_listen_fds(exec_cmdline, fd_array, m);
}
if (arg_quiet)
diff --git a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh
index dc5f952e59..7f888756ae 100755
--- a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh
+++ b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh
@@ -255,3 +255,120 @@ systemd-run --wait --pipe --user --machine testuser@ \
# test io.systemd.Unit in user manager
systemd-run --wait --pipe --user --machine testuser@ \
varlinkctl --more call "/run/user/$testuser_uid/systemd/io.systemd.Manager" io.systemd.Unit.List '{}'
+
+# test --upgrade (protocol upgrade)
+UPGRADE_SOCKET="$(mktemp -d)/upgrade.sock"
+UPGRADE_SERVER="$(mktemp)"
+cat >"$UPGRADE_SERVER" <<'PYEOF'
+#!/usr/bin/env python3
+"""Varlink upgrade test server. With a socket path argument, listens on a unix socket.
+Without arguments, speaks over stdin/stdout (for ssh-exec: transport testing)."""
+import json, os, socket, sys
+
+if len(sys.argv) > 1:
+ sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+ sock.bind(sys.argv[1])
+ sock.listen(1)
+ print("READY", flush=True)
+ conn, _ = sock.accept()
+ inp = conn.makefile("rb")
+ out = conn.makefile("wb")
+else:
+ inp = sys.stdin.buffer
+ out = sys.stdout.buffer
+ conn = sock = None
+
+# Read the varlink request (NUL-terminated JSON)
+data = b""
+while True:
+ chunk = inp.read(1)
+ assert chunk, "Connection closed before receiving full varlink request"
+ data += chunk
+ if b"\0" in data:
+ break
+
+msg = json.loads(data.split(b"\0")[0])
+received_parameters = msg.get("parameters", {})
+out.write(b'{"parameters": {}}\0')
+out.flush()
+
+# Upgraded protocol: send a non-JSON banner first to prove we're truly in raw mode,
+# then echo the received parameters, then reverse lines from the client
+out.write(b"<<< UPGRADED >>>\n")
+out.write((json.dumps(received_parameters) + "\n").encode())
+out.flush()
+for line in inp:
+ text = line.decode().rstrip("\n")
+ out.write((text[::-1] + "\n").encode())
+ out.flush()
+
+if conn:
+ conn.close()
+if sock:
+ sock.close()
+PYEOF
+chmod +x "$UPGRADE_SERVER"
+
+# Start the server in the background
+python3 "$UPGRADE_SERVER" "$UPGRADE_SOCKET" &
+SERVER_PID=$!
+
+# Wait for server readiness
+timeout 5 bash -c "while [ ! -S '$UPGRADE_SOCKET' ]; do sleep 0.1; done"
+
+# Test proxy mode: pipe data through --upgrade, passing parameters and validate
+result="$(echo "hello world" | varlinkctl call --upgrade "unix:$UPGRADE_SOCKET" io.systemd.test.Reverse '{"foo":"bar"}')"
+echo "$result" | grep "<<< UPGRADED >>>" >/dev/null
+echo "$result" | grep '"foo": "bar"' >/dev/null
+echo "$result" | grep "dlrow olleh" >/dev/null
+
+wait "$SERVER_PID" || :
+
+# Test --upgrade with stdin redirected from a regular file (epoll can't poll regular files,
+# so this exercises the fork+pipe fallback path)
+UPGRADE_SOCKET2="$(mktemp -d)/upgrade.sock"
+python3 "$UPGRADE_SERVER" "$UPGRADE_SOCKET2" &
+SERVER_PID=$!
+timeout 5 bash -c "while [ ! -S '$UPGRADE_SOCKET2' ]; do sleep 0.1; done"
+
+echo "file input test" > /tmp/test-upgrade-input
+result="$(varlinkctl call --upgrade "unix:$UPGRADE_SOCKET2" io.systemd.test.Reverse '{"foo":"file"}' < /tmp/test-upgrade-input)"
+echo "$result" | grep "<<< UPGRADED >>>" >/dev/null
+echo "$result" | grep '"foo": "file"' >/dev/null
+echo "$result" | grep "tset tupni elif" >/dev/null
+
+wait "$SERVER_PID" || :
+
+# Test --upgrade over ssh-exec: transport (pipe pair, not a bidirectional socket).
+# This exercises the input_fd != output_fd path in sd_varlink_call_and_upgrade().
+# Reuse the same server script without a socket argument - it speaks over stdin/stdout.
+cat > "$SSHBINDIR"/ssh <>>" >/dev/null
+echo "$result" | grep '"foo": "ssh"' >/dev/null
+echo "$result" | grep "tset epip hss" >/dev/null
+
+# Start another server for --exec test
+rm -f "$UPGRADE_SOCKET"
+python3 "$UPGRADE_SERVER" "$UPGRADE_SOCKET" &
+SERVER_PID=$!
+timeout 5 bash -c "while [ ! -S '$UPGRADE_SOCKET' ]; do sleep 0.1; done"
+
+# Test --exec mode: the upgraded socket becomes stdin/stdout of the child.
+# Since stdout goes to the socket (not the terminal), write results to a file for verification.
+EXEC_RESULT="$(mktemp)"
+varlinkctl call --upgrade --exec "unix:$UPGRADE_SOCKET" io.systemd.test.Reverse '{"foo":"bar"}' -- \
+ bash -c "head -2 > '$EXEC_RESULT'; echo 'hello world'; head -1 >> '$EXEC_RESULT'"
+grep "<<< UPGRADED >>>" "$EXEC_RESULT" >/dev/null
+grep '"foo": "bar"' "$EXEC_RESULT" >/dev/null
+grep "dlrow olleh" "$EXEC_RESULT" >/dev/null
+rm -f "$EXEC_RESULT"
+
+wait "$SERVER_PID" || :
+rm -f "$UPGRADE_SOCKET" "$UPGRADE_SOCKET2" "$UPGRADE_SERVER" /tmp/test-upgrade-input
+rm -rf "$(dirname "$UPGRADE_SOCKET")" "$(dirname "$UPGRADE_SOCKET2")"