From 79db432ab26231c24bf7da2a7517feb0a08c5914 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 24 Mar 2026 09:48:20 +0100 Subject: [PATCH 1/6] sd-varlink: add sd_varlink_call_and_upgrade() for protocol upgrades The varlink spec supports protocol upgrades and they are very useful to e.g. transfer binary data directly via varlink. So far sd-varlink was not supporting this. This commit adds a new public sd_varlink_call_and_upgrade() that sends a method call, waits for the reply, then steals the connection fds for raw I/O. It returns separate input_fd and output_fd to support both bidirectional sockets and pipe pairs. A helper is extracted and shared between sd_varlink_call_full() and sd_varlink_call_and_upgrade(). A new `protocol_upgrade` bool in `struct sd_varlink` ensures that on a protocol upgrade request we only exactly read the varlink protocol bytes and leave anything beyond that to the caller that speaks the upgraded protocol. Note that this is the client side of the library implementation only for now. The server side needs work but this is already useful as it allows to talk to varlink servers that speak protocol upgrades (like the rust implemenations of varlink). --- src/libsystemd/libsystemd.sym | 1 + src/libsystemd/sd-varlink/sd-varlink.c | 227 +++++++++++++++---- src/libsystemd/sd-varlink/varlink-internal.h | 2 + src/systemd/sd-varlink.h | 12 +- 4 files changed, 196 insertions(+), 46 deletions(-) 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/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); From 1af2d1c3e1e9546d76ac3e1eb70ffeb8dcf50441 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 24 Mar 2026 09:48:26 +0100 Subject: [PATCH 2/6] shared: extend socket-forward to support fd-pairs too Now that the socket forward code is extracted we can extend it to not just support bidirectional sockets but also input/output fd-pairs. This will be needed for e.g. the varlinkctl protocol upgrade support where one side of the connection is a fd-pair (stdin/stdout). This is done by creating two half-duplex forwarders that operate independantly. This also allows to simplify some state tracking, e.g. because each fd serves only one direction we don't need to dynamically create the event mask with EPOLLIN etc, its enough to set it once. It also handles non-pollable FDs transparently. Thanks to Lennart for his excellent suggestions here. --- src/shared/socket-forward.c | 348 ++++++++++++++++++++++++++---------- src/shared/socket-forward.h | 22 ++- 2 files changed, 267 insertions(+), 103 deletions(-) diff --git a/src/shared/socket-forward.c b/src/shared/socket-forward.c index 2601b25e6d..5ad8c5b2b6 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 { +/* Half-duplex forwarder: splices data from server_fd to client_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 server_to_client_buffer[2]; /* a pipe */ - int client_to_server_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 server_to_client_buffer_full, server_to_client_buffer_size; sd_event_source *server_event_source, *client_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->server_event_source); + sd_event_source_unref(fwd->client_event_source); - safe_close(sf->server_fd); - safe_close(sf->client_fd); + safe_close(fwd->server_fd); + safe_close(fwd->client_fd); - safe_close_pair(sf->server_to_client_buffer); - safe_close_pair(sf->client_to_server_buffer); + safe_close_pair(fwd->server_to_client_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->server_fd, fwd->server_to_client_buffer, &fwd->client_fd, + &fwd->server_to_client_buffer_full, &fwd->server_to_client_buffer_size, + &fwd->server_event_source, &fwd->client_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) - goto quit; - - /* EOF on both sides? */ - if (sf->server_fd < 0 && sf->client_fd < 0) - goto quit; - /* Server closed, and all data written to client? */ - if (sf->server_fd < 0 && sf->server_to_client_buffer_full <= 0) + if (fwd->server_fd < 0 && fwd->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) + /* Write side closed? */ + if (fwd->client_fd < 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->server_event_source = sd_event_source_disable_unref(fwd->server_event_source); + fwd->client_event_source = sd_event_source_disable_unref(fwd->client_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->server_to_client_buffer_full < fwd->server_to_client_buffer_size; + can_write = fwd->server_to_client_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->server_event_source) { + r = sd_event_source_set_enabled(fwd->server_event_source, can_read ? SD_EVENT_ONESHOT : SD_EVENT_OFF); + if (r < 0) + return log_debug_errno(r, "Failed to update server 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->client_event_source) { + r = sd_event_source_set_enabled(fwd->client_event_source, can_write ? SD_EVENT_ONESHOT : SD_EVENT_OFF); + if (r < 0) + return log_debug_errno(r, "Failed to update client 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 server_fd, + int client_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(server_fd >= 0); + assert(client_fd >= 0); + assert(server_fd != client_fd); + assert(on_done); + assert(ret); + + fwd = new(SimplexForward, 1); + if (!fwd) { + safe_close(server_fd); + safe_close(client_fd); + return log_oom_debug(); + } + + *fwd = (SimplexForward) { + .event = sd_event_ref(event), + .server_fd = server_fd, + .client_fd = client_fd, + .server_to_client_buffer = EBADF_PAIR, + .on_done = on_done, + .userdata = userdata, + }; + + r = socket_forward_create_pipes(fwd->server_to_client_buffer, &fwd->server_to_client_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->server_event_source, fwd->server_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->client_event_source, fwd->client_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->client_fd >= 0 && shutdown(fwd->client_fd, SHUT_WR) < 0) { + if (errno == ENOTSOCK) + fwd->client_fd = safe_close(fwd->client_fd); + else + log_debug_errno(errno, "Failed to shutdown write side of fd %d: %m, ignoring", + fwd->client_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); From 555dde10a059b095c3fcd1a679e8acc589182d33 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Thu, 26 Mar 2026 16:51:39 +0100 Subject: [PATCH 3/6] shared: rename internal variables in SimplexForwarder The SimplexForwader was using the naming of the SocketForwarder for bi-directional sockets. This was to keep the diff small and to make it easier to follow what changed and what was reused. However the name "client/server" for the SimplexForwader does no longer make much sense. The SimplexForwader is no longer about client/server but really just read/write. So this commit adjusts the naming. --- src/shared/socket-forward.c | 84 ++++++++++++++++++------------------- 1 file changed, 42 insertions(+), 42 deletions(-) diff --git a/src/shared/socket-forward.c b/src/shared/socket-forward.c index 5ad8c5b2b6..10c0a9be67 100644 --- a/src/shared/socket-forward.c +++ b/src/shared/socket-forward.c @@ -14,18 +14,18 @@ #define SOCKET_FORWARD_BUFFER_SIZE (256 * 1024) -/* Half-duplex forwarder: splices data from server_fd to client_fd via a kernel pipe buffer. +/* 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 buffer[2]; /* a pipe */ - size_t server_to_client_buffer_full, server_to_client_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; int (*on_done)(struct SimplexForward *fwd, int error, void *userdata); void *userdata; @@ -35,13 +35,13 @@ static SimplexForward* simplex_forward_free(SimplexForward *fwd) { if (!fwd) return NULL; - sd_event_source_unref(fwd->server_event_source); - sd_event_source_unref(fwd->client_event_source); + sd_event_source_unref(fwd->read_event_source); + sd_event_source_unref(fwd->write_event_source); - safe_close(fwd->server_fd); - safe_close(fwd->client_fd); + safe_close(fwd->read_fd); + safe_close(fwd->write_fd); - safe_close_pair(fwd->server_to_client_buffer); + safe_close_pair(fwd->buffer); sd_event_unref(fwd->event); @@ -131,18 +131,18 @@ static int simplex_forward_traffic(SimplexForward *fwd) { int r; r = simplex_forward_shovel( - &fwd->server_fd, fwd->server_to_client_buffer, &fwd->client_fd, - &fwd->server_to_client_buffer_full, &fwd->server_to_client_buffer_size, - &fwd->server_event_source, &fwd->client_event_source); + &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; - /* Server closed, and all data written to client? */ - if (fwd->server_fd < 0 && fwd->server_to_client_buffer_full <= 0) + /* Read side closed and all buffered data written? */ + if (fwd->read_fd < 0 && fwd->buffer_full <= 0) goto quit; /* Write side closed? */ - if (fwd->client_fd < 0) + if (fwd->write_fd < 0) goto quit; r = simplex_forward_enable_event_sources(fwd); @@ -152,8 +152,8 @@ static int simplex_forward_traffic(SimplexForward *fwd) { return 1; quit: - fwd->server_event_source = sd_event_source_disable_unref(fwd->server_event_source); - fwd->client_event_source = sd_event_source_disable_unref(fwd->client_event_source); + 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); } @@ -175,20 +175,20 @@ static int simplex_forward_enable_event_sources(SimplexForward *fwd) { assert(fwd); - can_read = fwd->server_to_client_buffer_full < fwd->server_to_client_buffer_size; - can_write = fwd->server_to_client_buffer_full > 0; + can_read = fwd->buffer_full < fwd->buffer_size; + can_write = fwd->buffer_full > 0; /* Event sources may have been unref'd by the shovel on EOF/disconnect */ - if (fwd->server_event_source) { - r = sd_event_source_set_enabled(fwd->server_event_source, can_read ? SD_EVENT_ONESHOT : SD_EVENT_OFF); + 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 server event source: %m"); + return log_debug_errno(r, "Failed to update read event source: %m"); } - if (fwd->client_event_source) { - r = sd_event_source_set_enabled(fwd->client_event_source, can_write ? SD_EVENT_ONESHOT : SD_EVENT_OFF); + 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 client event source: %m"); + return log_debug_errno(r, "Failed to update write event source: %m"); } return 0; @@ -214,8 +214,8 @@ static int simplex_forward_create_event_source( static int simplex_forward_new( sd_event *event, - int server_fd, - int client_fd, + int read_fd, + int write_fd, int (*on_done)(SimplexForward *fwd, int error, void *userdata), void *userdata, SimplexForward **ret) { @@ -224,37 +224,37 @@ static int simplex_forward_new( int r; assert(event); - assert(server_fd >= 0); - assert(client_fd >= 0); - assert(server_fd != client_fd); + 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(server_fd); - safe_close(client_fd); + safe_close(read_fd); + safe_close(write_fd); return log_oom_debug(); } *fwd = (SimplexForward) { .event = sd_event_ref(event), - .server_fd = server_fd, - .client_fd = client_fd, - .server_to_client_buffer = EBADF_PAIR, + .read_fd = read_fd, + .write_fd = write_fd, + .buffer = EBADF_PAIR, .on_done = on_done, .userdata = userdata, }; - r = socket_forward_create_pipes(fwd->server_to_client_buffer, &fwd->server_to_client_buffer_size); + r = socket_forward_create_pipes(fwd->buffer, &fwd->buffer_size); if (r < 0) return r; - r = simplex_forward_create_event_source(fwd, &fwd->server_event_source, fwd->server_fd, EPOLLIN); + r = simplex_forward_create_event_source(fwd, &fwd->read_event_source, fwd->read_fd, EPOLLIN); if (r < 0) return r; - r = simplex_forward_create_event_source(fwd, &fwd->client_event_source, fwd->client_fd, EPOLLOUT); + r = simplex_forward_create_event_source(fwd, &fwd->write_event_source, fwd->write_fd, EPOLLOUT); if (r < 0) return r; @@ -296,12 +296,12 @@ static int socket_forward_direction_done(SimplexForward *fwd, int error, void *u * (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->client_fd >= 0 && shutdown(fwd->client_fd, SHUT_WR) < 0) { + if (fwd->write_fd >= 0 && shutdown(fwd->write_fd, SHUT_WR) < 0) { if (errno == ENOTSOCK) - fwd->client_fd = safe_close(fwd->client_fd); + fwd->write_fd = safe_close(fwd->write_fd); else log_debug_errno(errno, "Failed to shutdown write side of fd %d: %m, ignoring", - fwd->client_fd); + fwd->write_fd); } if (error < 0 && sf->first_error >= 0) From 0fc3f85d35f155b1242423df120bdadecc0696e7 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 24 Mar 2026 09:48:33 +0100 Subject: [PATCH 4/6] varlinkctl: add protocol upgrade support The varlink spec supports protocol upgrades and they are very useful to e.g. transfer binary data directly via varlink. So far varlinkctl/sd-varlink was not supporting this. This commit adds support for it in varlinkctl by using the new code in sd-varlink and the generalized socket-forward code. --- man/varlinkctl.xml | 17 ++++ shell-completion/bash/varlinkctl | 2 +- shell-completion/zsh/_varlinkctl | 1 + src/varlinkctl/varlinkctl.c | 102 ++++++++++++++++++++- test/units/TEST-74-AUX-UTILS.varlinkctl.sh | 99 ++++++++++++++++++++ 5 files changed, 217 insertions(+), 4 deletions(-) diff --git a/man/varlinkctl.xml b/man/varlinkctl.xml index b96e35372f..2a771c1049 100644 --- a/man/varlinkctl.xml +++ b/man/varlinkctl.xml @@ -269,6 +269,23 @@ + + + + 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. Instead, 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. + + 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/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index 775a481ae8..e73c72ab97 100644 --- a/src/varlinkctl/varlinkctl.c +++ b/src/varlinkctl/varlinkctl.c @@ -30,6 +30,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 +53,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 +113,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 +143,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 +162,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 +251,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,6 +644,75 @@ 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); +} + +static int varlink_call_and_upgrade(const char *url, const char *method, sd_json_variant *parameters) { + _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"); + } + + /* 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; @@ -651,6 +730,12 @@ 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) { + /* TODO: support --exec with --upgrade, passing the upgraded socket fd to a child process */ + if (arg_exec || 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 --exec/--collect/--more/--oneway/--push-fd=/--graceful."); + } + (void) polkit_agent_open_if_enabled(BUS_TRANSPORT_LOCAL, arg_ask_password); url = argv[1]; @@ -658,6 +743,20 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) { parameter = argc > 3 && !streq(argv[3], "-") ? argv[3] : NULL; 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); + } + /* 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 * a user anyway) */ @@ -674,9 +773,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 = ""; diff --git a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh index dc5f952e59..24b29e55c3 100755 --- a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh +++ b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh @@ -255,3 +255,102 @@ 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() +data = inp.read().decode().rstrip("\n") +out.write((data[::-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 + +rm -f "$UPGRADE_SOCKET" "$UPGRADE_SOCKET2" "$UPGRADE_SERVER" /tmp/test-upgrade-input +rm -rf "$(dirname "$UPGRADE_SOCKET")" "$(dirname "$UPGRADE_SOCKET2")" From 042f633b12c72e20f532647655bbe4afcafc2aa0 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 24 Mar 2026 09:48:39 +0100 Subject: [PATCH 5/6] varlinkctl: add support for `--exec` with `--upgrade` Having support for `--exec` when using `--upgrade` is nice so this commit adds it. It does it by extracting a shared helper called `exec_with_listen_fds()` and then use that in the `verb_call()` and `varlink_call_and_upgrade()` calls. --- man/varlinkctl.xml | 17 +- src/varlinkctl/varlinkctl.c | 171 +++++++++++++-------- test/units/TEST-74-AUX-UTILS.varlinkctl.sh | 24 ++- 3 files changed, 139 insertions(+), 73 deletions(-) diff --git a/man/varlinkctl.xml b/man/varlinkctl.xml index 2a771c1049..6aff5a05e1 100644 --- a/man/varlinkctl.xml +++ b/man/varlinkctl.xml @@ -275,13 +275,20 @@ 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. Instead, 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. + 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 - . + , , or . diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index e73c72ab97..979797f8ec 100644 --- a/src/varlinkctl/varlinkctl.c +++ b/src/varlinkctl/varlinkctl.c @@ -650,7 +650,72 @@ static int upgrade_forward_done(SocketForward *sf, int error, void *userdata) { return sd_event_exit(event, error < 0 ? error : 0); } -static int varlink_call_and_upgrade(const char *url, const char *method, sd_json_variant *parameters) { +_noreturn_ static void exec_with_listen_fds(char **exec_cmdline, int *fds, size_t n_fds) { + int r; + + _cleanup_free_ char *j = strv_join(exec_cmdline, " "); + 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; @@ -684,6 +749,40 @@ static int varlink_call_and_upgrade(const char *url, const char *method, sd_json 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) @@ -717,7 +816,7 @@ 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); @@ -731,9 +830,8 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) { return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "--exec and --collect/--more/--oneway may not be combined."); if (arg_upgrade) { - /* TODO: support --exec with --upgrade, passing the upgraded socket fd to a child process */ - if (arg_exec || 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 --exec/--collect/--more/--oneway/--push-fd=/--graceful."); + 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); @@ -741,7 +839,7 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) { 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); @@ -754,7 +852,7 @@ static int verb_call(int argc, char *argv[], uintptr_t _data, void *userdata) { return log_error_errno(r, "Failed to parse parameters: %m"); } - return varlink_call_and_upgrade(url, method, jp); + 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 @@ -938,10 +1036,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); @@ -968,66 +1062,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 24b29e55c3..7f888756ae 100755 --- a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh +++ b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh @@ -297,9 +297,10 @@ out.flush() out.write(b"<<< UPGRADED >>>\n") out.write((json.dumps(received_parameters) + "\n").encode()) out.flush() -data = inp.read().decode().rstrip("\n") -out.write((data[::-1] + "\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() @@ -352,5 +353,22 @@ 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")" From 30b97055a5c7d81019a6c21bb63d9c033d4b8004 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Thu, 26 Mar 2026 16:32:48 +0100 Subject: [PATCH 6/6] varlink: tweak exec_with_listen_fds() log generation The previous code used strv_join() when it generated the log message for `varlinkctl --exec`. However this can lead to inaccurate logging so use `quote_command_line()` instead. Thanks to Lennart for suggesting this. --- src/varlinkctl/varlinkctl.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index 979797f8ec..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" @@ -653,7 +654,7 @@ static int upgrade_forward_done(SocketForward *sf, int error, void *userdata) { _noreturn_ static void exec_with_listen_fds(char **exec_cmdline, int *fds, size_t n_fds) { int r; - _cleanup_free_ char *j = strv_join(exec_cmdline, " "); + _cleanup_free_ char *j = quote_command_line(exec_cmdline, SHELL_ESCAPE_EMPTY); if (!j) { log_oom(); _exit(EXIT_FAILURE);