SymbX/systemd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

varlinkctl: add protocol upgrade support using new SocketFoward (#41283)

Browse Source
This commit is contained in:
Daan De Meyer
2026-04-01 12:33:00 +02:00
committed by GitHub
parent 51164fd9fc 30b97055a5
commit 8768af4019
11 changed files with 811 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files
man/varlinkctl.xml
+24
View File
@@ -269,6 +269,30 @@
<xi:include href="version-info.xml" xpointer="v255"/></listitem>
</varlistentry>
<varlistentry>
<term><option>--upgrade</option></term>
<listitem><para>When used with <command>call</command>: request a protocol upgrade. The method call
is sent with the <constant>upgrade</constant> 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.</para>
<para>If <option>--exec</option> is not specified, <command>varlinkctl</command> 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.</para>
<para>If <option>--exec</option> 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
<option>--exec</option> behavior (without <option>--upgrade</option>), 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.</para>
<para>This option may not be combined with <option>--more</option>, <option>--oneway</option>,
<option>--collect</option>, <option>--graceful=</option>, or <option>--push-fd=</option>.</para>
<xi:include href="version-info.xml" xpointer="v261"/></listitem>
</varlistentry>
<varlistentry>
<term><option>--json=<replaceable>MODE</replaceable></option></term>
shell-completion/bash/varlinkctl
+1 -1
View File
@@ -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'
)
shell-completion/zsh/_varlinkctl
+1
View File
@@ -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'
src/libsystemd/libsystemd.sym
+1
View File
@@ -1093,5 +1093,6 @@ global:
LIBSYSTEMD_261 {
global:
sd_varlink_call_and_upgrade;
sd_varlink_set_sentinel;
} LIBSYSTEMD_260;
src/libsystemd/sd-varlink/sd-varlink.c
+183 -44
View File
@@ -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,
src/libsystemd/sd-varlink/varlink-internal.h
+2
View File
@@ -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 */
src/shared/socket-forward.c
+253 -103
View File
@@ -1,6 +1,7 @@
/* SPDX-License-Identifier: LGPL-2.1-or-later */
#include <fcntl.h>
#include <sys/socket.h>
#include <unistd.h>
#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);
}
src/shared/socket-forward.h
+18 -4
View File
@@ -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);
src/systemd/sd-varlink.h
+10 -2
View File
@@ -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);
src/varlinkctl/varlinkctl.c
+201 -63
View File
@@ -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 = "<argv[4]>";
@@ -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)
test/units/TEST-74-AUX-UTILS.varlinkctl.sh
+117
View File
@@ -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 <<EOF
#!/usr/bin/env bash
exec python3 "$UPGRADE_SERVER"
EOF
chmod +x "$SSHBINDIR"/ssh
result="$(echo "ssh pipe test" | SYSTEMD_SSH="$SSHBINDIR/ssh" varlinkctl call --upgrade ssh-exec:foobar:test-upgrade io.systemd.test.Reverse '{"foo":"ssh"}')"
echo "$result" | grep "<<< UPGRADED >>>" >/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")"