From 82b04e7f8b7a5b4ee0cc50d92a830f8716ef2f77 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Thu, 2 Apr 2026 12:16:35 +0200 Subject: [PATCH 1/6] sd-varlink: extract varlink_handle_upgrade_fds() helper Extract the fd-handling logic from sd_varlink_call_and_upgrade() into a shared static helper so that it can be reused by the upcoming server-side sd_varlink_reply_and_upgrade(). --- src/libsystemd/sd-varlink/sd-varlink.c | 99 +++++++++++++++----------- 1 file changed, 57 insertions(+), 42 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 90be017705..25e67b4b74 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -2385,6 +2385,58 @@ _public_ int sd_varlink_call( return sd_varlink_call_full(v, method, parameters, ret_parameters, ret_error_id, NULL); } +static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret_output_fd) { + int r; + + assert(v); + assert(ret_input_fd || ret_output_fd); + + /* Ensure no post-upgrade data was consumed into our input buffer (we ensure this via MSG_PEEK or + * byte-to-byte) and refuse the upgrade rather than silently losing the data. */ + if (v->input_buffer_size != 0) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO), + "Unexpected buffered data during protocol upgrade, refusing."); + + /* 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) + return varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m"); + if (v->input_fd != v->output_fd) { + r = fd_nonblock(v->output_fd, false); + if (r < 0) + return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m"); + } + + /* Hand out the fds to the caller. When the caller doesn't want one direction, shut it + * down: but avoid closing the underlying fd if the other direction still needs it + * (i.e. when input_fd == output_fd). */ + bool same_fd = v->input_fd == v->output_fd; + + if (ret_input_fd) + *ret_input_fd = TAKE_FD(v->input_fd); + else { + (void) shutdown(v->input_fd, SHUT_RD); + if (same_fd && ret_output_fd) + TAKE_FD(v->input_fd); /* don't close yet, output branch needs it */ + else + v->input_fd = safe_close(v->input_fd); + } + + if (ret_output_fd) + *ret_output_fd = TAKE_FD(v->output_fd); + else { + (void) shutdown(v->output_fd, SHUT_WR); + if (same_fd && ret_input_fd) + TAKE_FD(v->output_fd); + else + v->output_fd = safe_close(v->output_fd); + } + + return 0; +} + _public_ int sd_varlink_call_and_upgrade( sd_varlink *v, const char *method, @@ -2436,45 +2488,12 @@ _public_ int sd_varlink_call_and_upgrade( 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); + /* Even if setting up the fds fails we must disconnect: the server already accepted the + * upgrade, so the other side is speaking raw protocol while we expect JSON. */ + r = varlink_handle_upgrade_fds(v, ret_input_fd, ret_output_fd); 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; - } - } - - /* Hand out the fds to the caller. When the caller doesn't want one direction, shut it - * down: but avoid closing the underlying fd if the other direction still needs it - * (i.e. when input_fd == output_fd). */ - bool same_fd = v->input_fd == v->output_fd; - - if (ret_input_fd) - *ret_input_fd = TAKE_FD(v->input_fd); - else { - (void) shutdown(v->input_fd, SHUT_RD); - if (same_fd && ret_output_fd) - TAKE_FD(v->input_fd); /* don't close yet, output branch needs it */ - else - v->input_fd = safe_close(v->input_fd); - } - - if (ret_output_fd) - *ret_output_fd = TAKE_FD(v->output_fd); - else { - (void) shutdown(v->output_fd, SHUT_WR); - if (same_fd && ret_input_fd) - TAKE_FD(v->output_fd); - else - v->output_fd = safe_close(v->output_fd); + varlink_set_state(v, VARLINK_DISCONNECTED); + goto finish; } varlink_set_state(v, VARLINK_DISCONNECTED); @@ -2488,10 +2507,6 @@ _public_ int sd_varlink_call_and_upgrade( 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); From 2c6f9af8e5425c2086fbc8ca496843f162e4af9b Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Wed, 1 Apr 2026 16:55:55 +0200 Subject: [PATCH 2/6] libsystemd: add sd_varlink_reply_and_upgrade protocol upgrade This commit adds protocol upgrade support in the libsystemd server side API code. --- src/libsystemd/libsystemd.sym | 1 + src/libsystemd/sd-varlink/sd-varlink-idl.c | 14 +++ src/libsystemd/sd-varlink/sd-varlink.c | 104 ++++++++++++++++++- src/systemd/sd-varlink-idl.h | 4 +- src/systemd/sd-varlink.h | 18 +++- src/test/test-varlink.c | 111 ++++++++++++++++++++- 6 files changed, 245 insertions(+), 7 deletions(-) diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym index 6af86aa2b4..619bcf820c 100644 --- a/src/libsystemd/libsystemd.sym +++ b/src/libsystemd/libsystemd.sym @@ -1094,5 +1094,6 @@ global: LIBSYSTEMD_261 { global: sd_varlink_call_and_upgrade; + sd_varlink_reply_and_upgrade; sd_varlink_set_sentinel; } LIBSYSTEMD_260; diff --git a/src/libsystemd/sd-varlink/sd-varlink-idl.c b/src/libsystemd/sd-varlink/sd-varlink-idl.c index 0b0ea244d6..dc09080cda 100644 --- a/src/libsystemd/sd-varlink/sd-varlink-idl.c +++ b/src/libsystemd/sd-varlink/sd-varlink-idl.c @@ -401,6 +401,16 @@ static int varlink_idl_format_symbol( fputs("\n", f); } + if ((symbol->symbol_flags & (SD_VARLINK_REQUIRES_UPGRADE|SD_VARLINK_SUPPORTS_UPGRADE)) != 0) { + fputs(colors[COLOR_COMMENT], f); + if (FLAGS_SET(symbol->symbol_flags, SD_VARLINK_REQUIRES_UPGRADE)) + fputs("# [Requires 'upgrade' flag]", f); + else + fputs("# [Supports 'upgrade' flag]", f); + fputs(colors[COLOR_RESET], f); + fputs("\n", f); + } + fputs(colors[COLOR_SYMBOL_TYPE], f); fputs("method ", f); fputs(colors[COLOR_IDENTIFIER], f); @@ -1945,6 +1955,10 @@ int varlink_idl_validate_method_call(const sd_varlink_symbol *method, sd_json_va if (FLAGS_SET(method->symbol_flags, SD_VARLINK_REQUIRES_MORE) && !FLAGS_SET(flags, SD_VARLINK_METHOD_MORE)) return -EBADE; + /* Same for upgrade */ + if (FLAGS_SET(method->symbol_flags, SD_VARLINK_REQUIRES_UPGRADE) && !FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE)) + return -EBADE; + return varlink_idl_validate_symbol(method, v, SD_VARLINK_INPUT, reterr_bad_field); } diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 25e67b4b74..1c03cfc173 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -1544,6 +1544,8 @@ static int varlink_dispatch_method(sd_varlink *v) { (flags & SD_VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY : VARLINK_PROCESSING_METHOD); + v->protocol_upgrade = FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE); + assert(v->server); /* First consult user supplied method implementations */ @@ -1566,11 +1568,15 @@ static int varlink_dispatch_method(sd_varlink *v) { r = varlink_idl_validate_method_call(v->current_method, parameters, flags, &bad_field); if (r == -EBADE) { - varlink_log_errno(v, r, "Method %s() called without 'more' flag, but flag needs to be set.", - method); + bool missing_upgrade = FLAGS_SET(v->current_method->symbol_flags, SD_VARLINK_REQUIRES_UPGRADE) && + !FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE); + + varlink_log_errno(v, r, "Method %s() called without '%s' flag, but flag needs to be set.", + method, missing_upgrade ? "upgrade" : "more"); if (v->state == VARLINK_PROCESSING_METHOD) { - r = sd_varlink_error(v, SD_VARLINK_ERROR_EXPECTED_MORE, NULL); + r = sd_varlink_error(v, missing_upgrade ? SD_VARLINK_ERROR_EXPECTED_UPGRADE + : SD_VARLINK_ERROR_EXPECTED_MORE, NULL); /* If we didn't manage to enqueue an error response, then fail the * connection completely. Otherwise ignore the error from * sd_varlink_error() here, as it is synthesized from the function's @@ -2821,6 +2827,97 @@ _public_ int sd_varlink_replyb(sd_varlink *v, ...) { return sd_varlink_reply(v, parameters); } +_public_ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parameters, int *ret_input_fd, int *ret_output_fd) { + int r; + + assert_return(v, -EINVAL); + assert_return(ret_input_fd || ret_output_fd, -EINVAL); + + if (v->state == VARLINK_DISCONNECTED) + return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); + + if (!IN_SET(v->state, + VARLINK_PROCESSING_METHOD, + VARLINK_PENDING_METHOD)) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy."); + + /* Verify the client actually requested a protocol upgrade */ + if (!v->protocol_upgrade) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO), + "Method call did not request a protocol upgrade."); + + /* Ensure we did not buffer any data beyond the upgrade request. Check this before sending the + * reply so that we can return a normal error (the framework will send an error reply to the + * client). In normal operation this cannot happen because the client waits for our reply before + * sending raw data, and we set protocol_upgrade=true in dispatch to limit subsequent reads to + * single bytes. But a misbehaving client could pipeline data early. */ + if (v->input_buffer_size > 0) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADMSG), + "Unexpected buffered data from client during protocol upgrade."); + + /* Validate parameters BEFORE sanitization (same validation as sd_varlink_reply(), but upgrade + * replies never carry the 'continues' flag so we always pass flags=0) */ + if (v->current_method) { + const char *bad_field = NULL; + + r = varlink_idl_validate_method_reply(v->current_method, parameters, /* flags= */ 0, &bad_field); + if (r < 0) + /* Please adjust test/units/end.sh when updating the log message. */ + varlink_log_errno(v, r, "Return parameters for method reply %s() didn't pass validation on field '%s', ignoring: %m", + v->current_method->name, strna(bad_field)); + } + + _cleanup_(sd_json_variant_unrefp) sd_json_variant *m = NULL; + r = sd_json_buildo(&m, 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); + if (r < 0) + return varlink_log_errno(v, r, "Failed to enqueue json message: %m"); + + /* Flush the reply to the socket before stealing the fds. The reply must be fully written + * before the caller starts speaking the upgraded protocol. */ + for (;;) { + r = varlink_write(v); + if (r < 0) { + varlink_log_errno(v, r, "Failed to flush reply: %m"); + goto disconnect; + } + if (v->output_buffer_size == 0 && !v->output_queue) + break; + if (v->write_disconnected) { + r = varlink_log_errno(v, SYNTHETIC_ERRNO(ECONNRESET), + "Write disconnected during upgrade reply flush."); + goto disconnect; + } + + r = fd_wait_for_event(v->output_fd, POLLOUT, USEC_INFINITY); + if (ERRNO_IS_NEG_TRANSIENT(r)) + continue; + if (r < 0) { + varlink_log_errno(v, r, "Failed to wait for writable fd: %m"); + goto disconnect; + } + assert(r > 0); + + handle_revents(v, r); + } + + /* Detach from the event loop before stealing the fds */ + varlink_detach_event_sources(v); + + /* Now hand the original FDs over to the caller, from this point on we have nothing to do with the + * connection anymore, it's up to the caller and we close the connection below */ + r = varlink_handle_upgrade_fds(v, ret_input_fd, ret_output_fd); + +disconnect: + /* This also sets the connection state to VARLINK_DISCONNECTED */ + sd_varlink_close(v); + + return r < 0 ? r : 1; +} + _public_ int sd_varlink_reset_fds(sd_varlink *v) { assert_return(v, -EINVAL); @@ -4572,6 +4669,7 @@ _public_ int sd_varlink_error_to_errno(const char *error, sd_json_variant *param { SD_VARLINK_ERROR_INVALID_PARAMETER, -EINVAL }, { SD_VARLINK_ERROR_PERMISSION_DENIED, -EACCES }, { SD_VARLINK_ERROR_EXPECTED_MORE, -EBADE }, + { SD_VARLINK_ERROR_EXPECTED_UPGRADE, -EPROTOTYPE }, }; int r; diff --git a/src/systemd/sd-varlink-idl.h b/src/systemd/sd-varlink-idl.h index ab85a95cc7..1122e31324 100644 --- a/src/systemd/sd-varlink-idl.h +++ b/src/systemd/sd-varlink-idl.h @@ -52,7 +52,9 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_symbol_type_t) { __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_symbol_flags_t) { SD_VARLINK_SUPPORTS_MORE = 1 << 0, /* Call supports "more" flag */ SD_VARLINK_REQUIRES_MORE = 1 << 1, /* Call requires "more" flag */ - _SD_VARLINK_SYMBOL_FLAGS_MAX = (1 << 2) - 1, + SD_VARLINK_SUPPORTS_UPGRADE = 1 << 2, /* Call supports "upgrade" flag */ + SD_VARLINK_REQUIRES_UPGRADE = 1 << 3, /* Call requires "upgrade" flag */ + _SD_VARLINK_SYMBOL_FLAGS_MAX = (1 << 4) - 1, _SD_VARLINK_SYMBOL_FLAGS_INVALID = -EINVAL, _SD_ENUM_FORCE_S64(SD_VARLINK_SYMBOL_FLAGS) } sd_varlink_symbol_flags_t; diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index 9d6e939d64..0b999b9154 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -137,8 +137,9 @@ int sd_varlink_callb(sd_varlink *v, const char *method, sd_json_variant **ret_pa 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. + * For bidirectional sockets ret_input_fd and ret_output_fd will be the same fd. Callers + * that need independent fds should dup() one of them. 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); @@ -168,6 +169,18 @@ int sd_varlink_replyb(sd_varlink *v, ...); #define sd_varlink_replybo(v, ...) \ sd_varlink_replyb((v), SD_JSON_BUILD_OBJECT(__VA_ARGS__)) +/* Send a final reply to an upgrade request, then steal the connection fds for raw I/O. + * The fds are returned in blocking mode. The varlink connection is disconnected afterwards. + * For bidirectional sockets ret_input_fd and ret_output_fd will be the same fd. Callers + * that need independent fds should dup() one of them. For pipe pairs (e.g. ssh-exec + * transport) they will differ. Either ret pointer may be NULL. + * + * Note: this call synchronously blocks until the reply is flushed to the socket. This is + * usually fine as flush is fast but a misbehaving/adversary client that stops reading + * could stall the caller. So do not use in servers that multiplex many varlink + * connections. */ +int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parameters, int *ret_input_fd, int *ret_output_fd); + /* Enqueue a (final) error */ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_variant *parameters); int sd_varlink_errorb(sd_varlink *v, const char *error_id, ...); @@ -322,6 +335,7 @@ _SD_DEFINE_POINTER_CLEANUP_FUNC(sd_varlink_server, sd_varlink_server_unref); #define SD_VARLINK_ERROR_INVALID_PARAMETER "org.varlink.service.InvalidParameter" #define SD_VARLINK_ERROR_PERMISSION_DENIED "org.varlink.service.PermissionDenied" #define SD_VARLINK_ERROR_EXPECTED_MORE "org.varlink.service.ExpectedMore" +#define SD_VARLINK_ERROR_EXPECTED_UPGRADE "org.varlink.service.ExpectedUpgrade" _SD_END_DECLARATIONS; diff --git a/src/test/test-varlink.c b/src/test/test-varlink.c index 3324421f68..36a4639376 100644 --- a/src/test/test-varlink.c +++ b/src/test/test-varlink.c @@ -11,6 +11,7 @@ #include "sd-varlink.h" #include "fd-util.h" +#include "io-util.h" #include "json-util.h" #include "memfd-util.h" #include "rm-rf.h" @@ -725,7 +726,7 @@ static int reply_notify_then_error(sd_varlink *link, sd_json_variant *parameters TEST(notify_then_error) { _cleanup_(sd_event_unrefp) sd_event *e = NULL; - ASSERT_OK(sd_event_default(&e)); + ASSERT_OK(sd_event_new(&e)); _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL; ASSERT_OK(sd_varlink_server_new(&s, 0)); @@ -752,4 +753,112 @@ TEST(notify_then_error) { ASSERT_OK(sd_event_loop(e)); } +static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { + _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF; + int r; + + ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE)); + + r = sd_varlink_reply_and_upgrade(link, /* parameters= */ NULL, &input_fd, &output_fd); + if (r < 0) + return r; + + /* For a socketpair connection, both fds point to the same socket — avoid double-close */ + if (input_fd == output_fd) + output_fd = -EBADF; + + /* After upgrade, do raw I/O: read until EOF, reverse, write back. + * The client shuts down its write side after sending, so we get a clean EOF. */ + char buf[64] = {}; + ssize_t n = ASSERT_OK(loop_read(input_fd, buf, sizeof(buf) - 1, /* do_poll= */ true)); + ASSERT_GT(n, 0); + + /* Reverse the received bytes */ + for (ssize_t i = 0; i < n / 2; i++) + SWAP_TWO(buf[i], buf[n - 1 - i]); + + int write_fd = output_fd >= 0 ? output_fd : input_fd; + ASSERT_OK(loop_write(write_fd, buf, n)); + + return 0; +} + +static int method_upgrade_without_flag(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { + int input_fd = -EBADF, output_fd = -EBADF; + + /* Calling reply_and_upgrade without the client requesting it should fail with -EPROTO */ + ASSERT_ERROR(sd_varlink_reply_and_upgrade(link, /* parameters= */ NULL, &input_fd, &output_fd), EPROTO); + + sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS); + + return sd_varlink_reply(link, /* parameters= */ NULL); +} + +static void *upgrade_thread(void *arg) { + _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *c = NULL; + _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF; + sd_json_variant *o = NULL; + const char *error_id = NULL; + + ASSERT_OK(sd_varlink_connect_address(&c, arg)); + ASSERT_OK(sd_varlink_set_description(c, "upgrade-client")); + + ASSERT_OK(sd_varlink_call_and_upgrade(c, "io.test.Upgrade", /* parameters= */ NULL, &o, &error_id, &input_fd, &output_fd)); + ASSERT_NULL(error_id); + ASSERT_GE(input_fd, 0); + ASSERT_GE(output_fd, 0); + + /* For a socketpair connection, both fds point to the same socket — avoid double-close */ + if (input_fd == output_fd) + output_fd = -EBADF; + + /* Send a test string, expect reversed reply */ + static const char msg[] = "Hello!"; + int write_fd = output_fd >= 0 ? output_fd : input_fd; + ASSERT_OK(loop_write(write_fd, msg, strlen(msg))); + ASSERT_OK_ERRNO(shutdown(write_fd, SHUT_WR)); + + char buf[64] = {}; + ssize_t n = ASSERT_OK(loop_read(input_fd, buf, strlen(msg), /* do_poll= */ true)); + ASSERT_EQ((size_t) n, strlen(msg)); + ASSERT_STREQ(buf, "!olleH"); + + /* Also test that a regular call (without upgrade flag) correctly rejects reply_and_upgrade on + * the server side, and still works as a normal call */ + _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *c2 = NULL; + ASSERT_OK(sd_varlink_connect_address(&c2, arg)); + ASSERT_OK(sd_varlink_set_description(c2, "no-upgrade-client")); + ASSERT_OK(sd_varlink_call(c2, "io.test.UpgradeWithoutFlag", /* parameters= */ NULL, &o, &error_id)); + ASSERT_NULL(error_id); + + return NULL; +} + +TEST(upgrade) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL; + _cleanup_(rm_rf_physical_and_freep) char *tmpdir = NULL; + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + pthread_t t; + const char *sp; + + ASSERT_OK(mkdtemp_malloc("/tmp/varlink-test-XXXXXX", &tmpdir)); + sp = strjoina(tmpdir, "/socket"); + + ASSERT_OK(sd_event_new(&e)); + + ASSERT_OK(sd_varlink_server_new(&s, 0)); + ASSERT_OK(sd_varlink_server_set_description(s, "upgrade-server")); + ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.Upgrade", method_upgrade)); + ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.UpgradeWithoutFlag", method_upgrade_without_flag)); + ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); + ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); + + ASSERT_OK(-pthread_create(&t, NULL, upgrade_thread, (void*) sp)); + + /* Run the event loop until no more connections (the thread will disconnect when done) */ + ASSERT_OK(sd_event_loop(e)); + + ASSERT_OK(-pthread_join(t, NULL)); +} + DEFINE_TEST_MAIN(LOG_DEBUG); From 34f29079fdba7eb3820e6e79e370671fd293bd87 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Thu, 2 Apr 2026 09:38:41 +0200 Subject: [PATCH 3/6] varlinkctl: add new `serve` verb to allow wrapping command in varlink With the new protocol upgrade support in varlinkctl client we can now do the equivalent for the server side. This commit adds a new `serve` verb that will serve any command that speaks stdin/stdout via varlink and its protocol upgrade feature. This is the "inetd for varlink". This is useful for various reasons: 1. Allows to e.g. provide a heavily sandboxed io.myorg.xz.Decompress varlink endpoint, c.f. xz CVE-2024-3094) 2. Allow sftp over varlink which is quite useful with the varlink-http-bridge (that has more flexible auth mechanism than plain sftp). 3. Makes testing the varlinkctl client protocol upgrade simpler. 4. Because we can. --- man/varlinkctl.xml | 70 ++++++++++ src/varlinkctl/varlinkctl.c | 152 +++++++++++++++++++++ test/units/TEST-74-AUX-UTILS.varlinkctl.sh | 48 +++++-- 3 files changed, 261 insertions(+), 9 deletions(-) diff --git a/man/varlinkctl.xml b/man/varlinkctl.xml index 6aff5a05e1..adf26b8fe6 100644 --- a/man/varlinkctl.xml +++ b/man/varlinkctl.xml @@ -73,6 +73,14 @@ CMDLINE + + varlinkctl + OPTIONS + serve + METHOD + CMDLINE + + varlinkctl OPTIONS @@ -181,6 +189,28 @@ + + serve METHOD CMDLINE… + + Run a Varlink server that accepts protocol upgrade requests for the specified method + and connects the upgraded connection to the standard input and output of the specified command. This + can act as a server-side counterpart to call . + + The listening socket must be passed via socket activation (i.e. the + $LISTEN_FDS protocol), making this command suitable for use in socket-activated + service units. When a client calls the specified method with the upgrade flag, the server sends a + reply confirming the upgrade, then forks and executes the given command line with the upgraded + connection on its standard input and output. + + This effectively turns any command that speaks a protocol over standard input/output into a + Varlink service, discoverable via the service registry and authenticated via socket credentials. + Because each connection is handled by a forked child process, the service unit can apply systemd's + sandboxing options (such as ProtectSystem=, etc.) and does not operate in the + caller's environment. + + + + list-registry @@ -533,6 +563,46 @@ method Extend( # varlinkctl call ssh-exec:somehost:systemd-creds org.varlink.service.GetInfo '{}' + + Serving a Sandboxed Decompressor via Protocol Upgrade + + The following socket and service units expose xz decompression as a Varlink + service. Clients connect and send compressed data over the upgraded connection, receiving decompressed + output in return. + + # /etc/systemd/system/varlink-decompress-xz.socket +[Socket] +ListenStream=/run/varlink/registry/com.example.Decompress.XZ + +[Install] +WantedBy=sockets.target + +# /etc/systemd/system/varlink-decompress-xz.service +[Service] +ExecStart=varlinkctl serve com.example.Decompress.XZ xz -d +DynamicUser=yes +PrivateNetwork=yes +ProtectSystem=strict +ProtectHome=yes +NoNewPrivileges=yes +SystemCallFilter=~@privileged @resources +MemoryMax=256M + + A client can then decompress data through this service: + + $ echo "hello" | xz | varlinkctl call --upgrade \ + unix:/run/varlink/registry/com.example.Decompress.XZ \ + com.example.Decompress.XZ '{}' +hello + + For quick testing without unit files, systemd-socket-activate can be used + to provide the listening socket: + + $ systemd-socket-activate -l /tmp/decompress.sock -- varlinkctl serve com.example.Decompress.XZ xz -d & +$ echo "hello" | xz | varlinkctl call --upgrade unix:/tmp/decompress.sock com.example.Decompress.XZ '{}' +hello + + diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index 00bd71f34d..1876b90bde 100644 --- a/src/varlinkctl/varlinkctl.c +++ b/src/varlinkctl/varlinkctl.c @@ -1169,6 +1169,158 @@ static int verb_list_registry(int argc, char *argv[], uintptr_t _data, void *use return 0; } +/* Build a minimal IDL from a qualified method name so that introspection works. The parsed interface is + * returned to the caller who must keep it alive for the lifetime of the server + * (sd_varlink_server_add_interface() borrows the pointer). */ +static int varlink_server_add_interface_from_method(sd_varlink_server *s, const char *method, sd_varlink_interface **ret_interface) { + assert(s); + assert(method); + assert(ret_interface); + + const char *dot = strrchr(method, '.'); + assert(dot); + + _cleanup_free_ char *interface_name = strndup(method, dot - method); + if (!interface_name) + return log_oom(); + + /* Note that we do not need to put the upgrade flag comment here, it is added automatically + * by varlink_idl_format_symbol() because of the SD_VARLINK_REQUIRES_UPGRADE flag. */ + _cleanup_free_ char *idl_text = strjoin( + "interface ", interface_name, "\n" + "\n" + "method ", dot + 1, " () -> ()\n"); + if (!idl_text) + return log_oom(); + + _cleanup_(sd_varlink_interface_freep) sd_varlink_interface *iface = NULL; + int r = sd_varlink_idl_parse(idl_text, /* reterr_line= */ NULL, /* reterr_column= */ NULL, &iface); + if (r < 0) + return log_error_errno(r, "Failed to parse IDL for method '%s': %m", method); + + /* Mark the method as requiring the upgrade flag so introspection shows the annotation */ + assert(iface->symbols[0] && iface->symbols[0]->symbol_type == SD_VARLINK_METHOD); + ((sd_varlink_symbol*) iface->symbols[0])->symbol_flags |= SD_VARLINK_REQUIRES_UPGRADE; + + r = sd_varlink_server_add_interface(s, iface); + if (r < 0) + return r; + + *ret_interface = TAKE_PTR(iface); + + return 0; +} + +static int method_serve_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { + char **exec_cmdline = ASSERT_PTR(userdata); + _cleanup_close_ int input_fd = -EBADF, _output_fd = -EBADF; + int output_fd, r; + + if (!FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE)) + return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_UPGRADE, NULL); + + r = sd_varlink_reply_and_upgrade(link, /* parameters= */ NULL, &input_fd, &output_fd); + if (r < 0) + return log_error_errno(r, "Failed to upgrade connection: %m"); + + if (output_fd != input_fd) + _output_fd = output_fd; + + /* Copy exec_cmdline before forking: pidref_safe_fork() calls rename_process() which + * overwrites the argv area that exec_cmdline points into. */ + _cleanup_strv_free_ char **cmdline_copy = strv_copy(exec_cmdline); + if (!cmdline_copy) + return log_oom(); + + r = pidref_safe_fork_full( + "(serve)", + (int[]) { input_fd, output_fd, STDERR_FILENO }, + /* except_fds= */ NULL, /* n_except_fds= */ 0, + FORK_RESET_SIGNALS|FORK_CLOSE_ALL_FDS|FORK_REARRANGE_STDIO|FORK_DETACH|FORK_LOG, + /* ret= */ NULL); + if (r < 0) + return r; + if (r == 0) { + execvp(cmdline_copy[0], cmdline_copy); + log_error_errno(errno, "Failed to execute '%s': %m", cmdline_copy[0]); + _exit(EXIT_FAILURE); + } + + return 0; +} + +VERB(verb_serve, "serve", "METHOD CMDLINE…", 3, VERB_ANY, 0, "Serve a command via varlink protocol upgrade"); +static int verb_serve(int argc, char *argv[], uintptr_t _data, void *userdata) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL; + _cleanup_(sd_event_unrefp) sd_event *event = NULL; + const char *method; + char **exec_cmdline; + int r, n; + + assert(argc >= 3); /* Guaranteed by verb dispatch table */ + + method = argv[1]; + exec_cmdline = argv + 2; + + r = varlink_idl_qualified_symbol_name_is_valid(method); + if (r < 0) + return log_error_errno(r, "Failed to validate method name '%s': %m", method); + if (r == 0) + return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Not a valid qualified method name: '%s'", method); + + /* Require socket activation */ + n = sd_listen_fds(/* unset_environment= */ true); + if (n < 0) + return log_error_errno(n, "Failed to determine passed file descriptors: %m"); + if (n == 0) + return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "No file descriptors passed via socket activation."); + if (n > 1) + return log_error_errno(SYNTHETIC_ERRNO(EINVAL), "Expected exactly one socket activation fd, got %d.", n); + + r = sd_event_default(&event); + if (r < 0) + return log_error_errno(r, "Failed to get event loop: %m"); + + r = sd_varlink_server_new(&s, SD_VARLINK_SERVER_INHERIT_USERDATA); + if (r < 0) + return log_error_errno(r, "Failed to allocate varlink server: %m"); + + _cleanup_free_ char *description = strjoin("serve:", method); + if (!description) + return log_oom(); + + r = sd_varlink_server_set_description(s, description); + if (r < 0) + return log_error_errno(r, "Failed to set server description: %m"); + + r = sd_varlink_server_bind_method(s, method, method_serve_upgrade); + if (r < 0) + return log_error_errno(r, "Failed to bind method '%s': %m", method); + + _cleanup_(sd_varlink_interface_freep) sd_varlink_interface *iface = NULL; + r = varlink_server_add_interface_from_method(s, method, &iface); + if (r < 0) + return log_error_errno(r, "Failed to add interface for method '%s': %m", method); + + sd_varlink_server_set_userdata(s, exec_cmdline); + + r = sd_varlink_server_listen_fd(s, SD_LISTEN_FDS_START); + if (r < 0) + return log_error_errno(r, "Failed to listen on socket activation fd: %m"); + + r = sd_varlink_server_attach_event(s, event, SD_EVENT_PRIORITY_NORMAL); + if (r < 0) + return log_error_errno(r, "Failed to attach varlink server to event loop: %m"); + + (void) sd_notify(/* unset_environment= */ false, "READY=1"); + + r = sd_event_loop(event); + if (r < 0) + return log_error_errno(r, "Failed to run event loop: %m"); + + return 0; +} + static int run(int argc, char *argv[]) { int r; diff --git a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh index b6d270cfd4..782a6dc6e9 100755 --- a/test/units/TEST-74-AUX-UTILS.varlinkctl.sh +++ b/test/units/TEST-74-AUX-UTILS.varlinkctl.sh @@ -257,6 +257,9 @@ 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) +# The basic --upgrade proxy test is covered by the "varlinkctl serve" tests below (which use +# serve+rev/gunzip as the server). The tests here exercise features that need the Python +# server: file-input (defer fallback), ssh-exec transport (pipe pairs) and --exec mode. UPGRADE_SOCKET="$(mktemp -d)/upgrade.sock" UPGRADE_SERVER="$(mktemp)" cat >"$UPGRADE_SERVER" <<'PYEOF' @@ -320,15 +323,6 @@ if sock: PYEOF chmod +x "$UPGRADE_SERVER" -# Start the server in the background, wait for readiness via sd_notify -systemd-notify --fork -q -- python3 "$UPGRADE_SERVER" "$UPGRADE_SOCKET" - -# 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 - # Test --upgrade with stdin redirected from a regular file (epoll can't poll regular files, # so this exercises the sd_event_add_defer fallback path) UPGRADE_SOCKET2="$(mktemp -d)/upgrade.sock" @@ -370,3 +364,39 @@ rm -f "$EXEC_RESULT" rm -f "$UPGRADE_SOCKET" "$UPGRADE_SOCKET2" "$UPGRADE_SERVER" /tmp/test-upgrade-input rm -rf "$(dirname "$UPGRADE_SOCKET")" "$(dirname "$UPGRADE_SOCKET2")" + +# Test varlinkctl serve: expose a stdio command via varlink protocol upgrade with socket activation. +# This is the "inetd for varlink" pattern: any stdio tool becomes a varlink service. +SERVE_SOCKET="$(mktemp -d)/serve.sock" + +# Test 1: serve rev: proves bidirectional data flow through the upgrade +SERVE_PID=$(systemd-notify --fork -- \ + systemd-socket-activate -l "$SERVE_SOCKET" -- \ + varlinkctl serve io.systemd.test.Reverse rev) + +# Verify introspection works on the serve endpoint and shows the upgrade annotation +varlinkctl introspect "unix:$SERVE_SOCKET" io.systemd.test | grep "method Reverse" >/dev/null +varlinkctl introspect "unix:$SERVE_SOCKET" io.systemd.test | grep "Requires 'upgrade' flag" >/dev/null + +result="$(echo "hello world" | varlinkctl call --upgrade "unix:$SERVE_SOCKET" io.systemd.test.Reverse '{}')" +echo "$result" | grep "dlrow olleh" >/dev/null +kill "$SERVE_PID" 2>/dev/null || true +wait "$SERVE_PID" 2>/dev/null || true +rm -f "$SERVE_SOCKET" + +# Test 2: decompress via serve: the "sandboxed decompressor" use-case (the real thing would be a proper +# unit with real sandboxing). +# Pipe gzip-compressed data through a varlinkctl serve + gunzip endpoint and verify round-trip. +SERVE_PID=$(systemd-notify --fork -- \ + systemd-socket-activate -l "$SERVE_SOCKET" -- \ + varlinkctl serve io.systemd.Compress.Decompress gunzip) + +SERVE_TMPDIR="$(mktemp -d)" +echo "untrusted data decompressed safely via varlink serve" | gzip > "$SERVE_TMPDIR/compressed.gz" +result="$(varlinkctl call --upgrade "unix:$SERVE_SOCKET" io.systemd.Compress.Decompress '{}' < "$SERVE_TMPDIR/compressed.gz")" +echo "$result" | grep "untrusted data decompressed safely" >/dev/null +kill "$SERVE_PID" 2>/dev/null || true +wait "$SERVE_PID" 2>/dev/null || true + +rm -f "$SERVE_SOCKET" +rm -rf "$(dirname "$SERVE_SOCKET")" "$SERVE_TMPDIR" From 0d21b105d23d9015f87121a2ec6383aa9c9df421 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Sun, 5 Apr 2026 10:05:30 +0200 Subject: [PATCH 4/6] libsystemd,varlink: always return two fds in varlink upgrade API This commit tweaks the API of sd_varlink_call_and_upgrade and sd_varlink_reply_and_upgrade to return two independent fds even if the internal {input,output}_fd are the same (e.g. a socket). This makes the external API easier as there is no longer the risk of double close. The sd_varlink_call_and_upgrade() is not in a released version of systemd yet so I presume it is okay to update it still. This also allowed some simplifications in varlinkctl.c now that the handling is easier. --- src/libsystemd/sd-varlink/sd-varlink.c | 22 ++++++++++------------ src/systemd/sd-varlink.h | 12 ++++++------ src/test/test-varlink.c | 19 +++++-------------- src/varlinkctl/varlinkctl.c | 16 ++-------------- 4 files changed, 23 insertions(+), 46 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 1c03cfc173..fcf1704792 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -2415,29 +2415,27 @@ static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m"); } - /* Hand out the fds to the caller. When the caller doesn't want one direction, shut it - * down: but avoid closing the underlying fd if the other direction still needs it - * (i.e. when input_fd == output_fd). */ - bool same_fd = v->input_fd == v->output_fd; + /* For bidirectional sockets (input_fd == output_fd), dup the fd so that callers + * always get two independent fds they can close separately. */ + if (v->input_fd == v->output_fd) { + v->output_fd = fcntl(v->input_fd, F_DUPFD_CLOEXEC, 3); + if (v->output_fd < 0) + return varlink_log_errno(v, errno, "Failed to dup upgraded connection fd: %m"); + } + /* Hand out requested fds, shut down unwanted directions. */ if (ret_input_fd) *ret_input_fd = TAKE_FD(v->input_fd); else { (void) shutdown(v->input_fd, SHUT_RD); - if (same_fd && ret_output_fd) - TAKE_FD(v->input_fd); /* don't close yet, output branch needs it */ - else - v->input_fd = safe_close(v->input_fd); + v->input_fd = safe_close(v->input_fd); } if (ret_output_fd) *ret_output_fd = TAKE_FD(v->output_fd); else { (void) shutdown(v->output_fd, SHUT_WR); - if (same_fd && ret_input_fd) - TAKE_FD(v->output_fd); - else - v->output_fd = safe_close(v->output_fd); + v->output_fd = safe_close(v->output_fd); } return 0; diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index 0b999b9154..527415ab80 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -137,9 +137,9 @@ int sd_varlink_callb(sd_varlink *v, const char *method, sd_json_variant **ret_pa 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. Callers - * that need independent fds should dup() one of them. ret_parameters and ret_error_id are - * borrowed references valid only until v is closed or unreffed. + * For bidirectional sockets ret_input_fd and ret_output_fd will be separate (dupped) fds + * referring to the same underlying socket. 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); @@ -171,9 +171,9 @@ int sd_varlink_replyb(sd_varlink *v, ...); /* Send a final reply to an upgrade request, then steal the connection fds for raw I/O. * The fds are returned in blocking mode. The varlink connection is disconnected afterwards. - * For bidirectional sockets ret_input_fd and ret_output_fd will be the same fd. Callers - * that need independent fds should dup() one of them. For pipe pairs (e.g. ssh-exec - * transport) they will differ. Either ret pointer may be NULL. + * For bidirectional sockets ret_input_fd and ret_output_fd will be separate (dupped) fds + * referring to the same underlying socket. For pipe pairs (e.g. ssh-exec transport) they + * will differ. Either ret pointer may be NULL. * * Note: this call synchronously blocks until the reply is flushed to the socket. This is * usually fine as flush is fast but a misbehaving/adversary client that stops reading diff --git a/src/test/test-varlink.c b/src/test/test-varlink.c index 36a4639376..219966c5d7 100644 --- a/src/test/test-varlink.c +++ b/src/test/test-varlink.c @@ -763,10 +763,6 @@ static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varl if (r < 0) return r; - /* For a socketpair connection, both fds point to the same socket — avoid double-close */ - if (input_fd == output_fd) - output_fd = -EBADF; - /* After upgrade, do raw I/O: read until EOF, reverse, write back. * The client shuts down its write side after sending, so we get a clean EOF. */ char buf[64] = {}; @@ -777,8 +773,7 @@ static int method_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varl for (ssize_t i = 0; i < n / 2; i++) SWAP_TWO(buf[i], buf[n - 1 - i]); - int write_fd = output_fd >= 0 ? output_fd : input_fd; - ASSERT_OK(loop_write(write_fd, buf, n)); + ASSERT_OK(loop_write(output_fd, buf, n)); return 0; } @@ -807,16 +802,12 @@ static void *upgrade_thread(void *arg) { ASSERT_NULL(error_id); ASSERT_GE(input_fd, 0); ASSERT_GE(output_fd, 0); + ASSERT_NE(input_fd, output_fd); /* library dups for bidirectional sockets */ - /* For a socketpair connection, both fds point to the same socket — avoid double-close */ - if (input_fd == output_fd) - output_fd = -EBADF; - - /* Send a test string, expect reversed reply */ + /* Send a test string, shut down write side so server sees EOF, then read the reversed reply */ static const char msg[] = "Hello!"; - int write_fd = output_fd >= 0 ? output_fd : input_fd; - ASSERT_OK(loop_write(write_fd, msg, strlen(msg))); - ASSERT_OK_ERRNO(shutdown(write_fd, SHUT_WR)); + ASSERT_OK(loop_write(output_fd, msg, strlen(msg))); + ASSERT_OK_ERRNO(shutdown(output_fd, SHUT_WR)); char buf[64] = {}; ssize_t n = ASSERT_OK(loop_read(input_fd, buf, strlen(msg), /* do_poll= */ true)); diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index 1876b90bde..9a8e89b26c 100644 --- a/src/varlinkctl/varlinkctl.c +++ b/src/varlinkctl/varlinkctl.c @@ -672,15 +672,6 @@ static int varlink_call_and_upgrade(const char *url, const char *method, sd_json 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. */ @@ -1213,8 +1204,8 @@ static int varlink_server_add_interface_from_method(sd_varlink_server *s, const static int method_serve_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { char **exec_cmdline = ASSERT_PTR(userdata); - _cleanup_close_ int input_fd = -EBADF, _output_fd = -EBADF; - int output_fd, r; + _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF; + int r; if (!FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE)) return sd_varlink_error(link, SD_VARLINK_ERROR_EXPECTED_UPGRADE, NULL); @@ -1223,9 +1214,6 @@ static int method_serve_upgrade(sd_varlink *link, sd_json_variant *parameters, s if (r < 0) return log_error_errno(r, "Failed to upgrade connection: %m"); - if (output_fd != input_fd) - _output_fd = output_fd; - /* Copy exec_cmdline before forking: pidref_safe_fork() calls rename_process() which * overwrites the argv area that exec_cmdline points into. */ _cleanup_strv_free_ char **cmdline_copy = strv_copy(exec_cmdline); From 34b9607e4e7621e93a8d3418799eac3e16be5d30 Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 7 Apr 2026 17:47:50 +0200 Subject: [PATCH 5/6] varlink: use single byte reads on SD_VARLINK_SERVER_UPGRADABLE When the server side of a varlink connection supports connection upgrades we need to go into single byte-read mode to avoid the risk of a client that sends the json to protocol upgrade and then immediately the custom protocol payload. This commit implements this. The next step is using MSG_PEEK to avoid the single-byte overhead. --- src/libsystemd/sd-varlink/sd-varlink.c | 3 +- src/systemd/sd-varlink.h | 1 + src/test/test-varlink.c | 86 +++++++++++++++++++++++++- src/varlinkctl/varlinkctl.c | 2 +- 4 files changed, 89 insertions(+), 3 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index fcf1704792..bc7e93f407 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -3710,7 +3710,8 @@ _public_ int sd_varlink_server_new(sd_varlink_server **ret, sd_varlink_server_fl SD_VARLINK_SERVER_ALLOW_FD_PASSING_OUTPUT| SD_VARLINK_SERVER_FD_PASSING_INPUT_STRICT| SD_VARLINK_SERVER_HANDLE_SIGINT| - SD_VARLINK_SERVER_HANDLE_SIGTERM)) == 0, -EINVAL); + SD_VARLINK_SERVER_HANDLE_SIGTERM| + SD_VARLINK_SERVER_UPGRADABLE)) == 0, -EINVAL); s = new(sd_varlink_server, 1); if (!s) diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index 527415ab80..3be82a7ddb 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -72,6 +72,7 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_server_flags_t) { SD_VARLINK_SERVER_FD_PASSING_INPUT_STRICT = 1 << 7, /* Reject input messages with fds if fd passing is disabled (needs kernel v6.16+) */ SD_VARLINK_SERVER_HANDLE_SIGINT = 1 << 8, /* Exit cleanly on SIGINT */ SD_VARLINK_SERVER_HANDLE_SIGTERM = 1 << 9, /* Exit cleanly on SIGTERM */ + SD_VARLINK_SERVER_UPGRADABLE = 1 << 10, /* Server has upgrade methods; avoid consuming post-upgrade data during reads */ _SD_ENUM_FORCE_S64(SD_VARLINK_SERVER) } sd_varlink_server_flags_t; diff --git a/src/test/test-varlink.c b/src/test/test-varlink.c index 219966c5d7..1bbc87c32c 100644 --- a/src/test/test-varlink.c +++ b/src/test/test-varlink.c @@ -15,6 +15,7 @@ #include "json-util.h" #include "memfd-util.h" #include "rm-rf.h" +#include "socket-util.h" #include "tests.h" #include "tmpfile-util.h" #include "varlink-util.h" @@ -837,7 +838,7 @@ TEST(upgrade) { ASSERT_OK(sd_event_new(&e)); - ASSERT_OK(sd_varlink_server_new(&s, 0)); + ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_UPGRADABLE)); ASSERT_OK(sd_varlink_server_set_description(s, "upgrade-server")); ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.Upgrade", method_upgrade)); ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.UpgradeWithoutFlag", method_upgrade_without_flag)); @@ -852,4 +853,87 @@ TEST(upgrade) { ASSERT_OK(-pthread_join(t, NULL)); } +static int method_upgrade_and_exit(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { + sd_event *event = ASSERT_PTR(userdata); + + int r = method_upgrade(link, parameters, flags, /* userdata= */ NULL); + + /* Exit the event loop after the upgrade is handled. We can't use sd_varlink_get_event() + * here because the connection is already disconnected after reply_and_upgrade. */ + (void) sd_event_exit(event, r < 0 ? r : EXIT_SUCCESS); + return r; +} + +static void *upgrade_pipelining_thread(void *arg) { + union sockaddr_union sa = {}; + _cleanup_close_ int fd = -EBADF; + + /* Connect a raw socket and pipeline: upgrade JSON + \0 + raw data in a single write. + * This tests that the server's byte-by-byte reading (SD_VARLINK_SERVER_UPGRADABLE) + * doesn't consume the raw data into the varlink input buffer. */ + fd = socket(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0); + ASSERT_FD(fd); + int addrlen = sockaddr_un_set_path(&sa.un, arg); + ASSERT_OK(addrlen); + ASSERT_OK_ERRNO(connect(fd, &sa.sa, addrlen)); + + /* Build pipelined message: upgrade JSON + \0 + raw payload, all in one write */ + static const char upgrade_msg[] = "{\"method\":\"io.test.Upgrade\",\"upgrade\":true}"; + static const char raw_payload[] = "Pipelined!"; + char send_buf[sizeof(upgrade_msg) + sizeof(raw_payload)]; /* includes \0 from upgrade_msg as delimiter */ + + memcpy(send_buf, upgrade_msg, sizeof(upgrade_msg)); /* copies trailing \0 = varlink delimiter */ + memcpy(send_buf + sizeof(upgrade_msg), raw_payload, sizeof(raw_payload) - 1); + + size_t total = sizeof(upgrade_msg) + strlen(raw_payload); + ASSERT_OK(loop_write(fd, send_buf, total)); + + /* Shut down write side so server's method_upgrade sees EOF after raw payload */ + ASSERT_OK_ERRNO(shutdown(fd, SHUT_WR)); + + /* Read everything: upgrade reply (JSON + \0) + reversed raw payload. The server closes + * the connection after writing, so loop_read() reads until EOF and gets it all. */ + char buf[256] = {}; + ssize_t n = ASSERT_OK(loop_read(fd, buf, sizeof(buf) - 1, /* do_poll= */ true)); + ASSERT_GT(n, 0); + + /* Split at the \0 delimiter between JSON reply and raw payload */ + char *delim = memchr(buf, 0, n); + ASSERT_NOT_NULL(delim); + + char *raw = delim + 1; + size_t raw_size = (size_t) n - (size_t)(raw - buf); + + ASSERT_EQ(raw_size, strlen(raw_payload)); + ASSERT_STREQ(strndupa_safe(raw, raw_size), "!denilepiP"); + + return NULL; +} + +TEST(upgrade_pipelining) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL; + _cleanup_(rm_rf_physical_and_freep) char *tmpdir = NULL; + _cleanup_(sd_event_unrefp) sd_event *e = NULL; + pthread_t t; + const char *sp; + + ASSERT_OK(mkdtemp_malloc("/tmp/varlink-test-XXXXXX", &tmpdir)); + sp = strjoina(tmpdir, "/socket"); + + ASSERT_OK(sd_event_new(&e)); + + ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_UPGRADABLE|SD_VARLINK_SERVER_INHERIT_USERDATA)); + ASSERT_OK(sd_varlink_server_set_description(s, "upgrade-pipelining-server")); + ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.Upgrade", method_upgrade_and_exit)); + ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); + ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); + sd_varlink_server_set_userdata(s, e); + + ASSERT_OK(-pthread_create(&t, NULL, upgrade_pipelining_thread, (void*) sp)); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_OK(-pthread_join(t, NULL)); +} + DEFINE_TEST_MAIN(LOG_DEBUG); diff --git a/src/varlinkctl/varlinkctl.c b/src/varlinkctl/varlinkctl.c index 9a8e89b26c..18a639962c 100644 --- a/src/varlinkctl/varlinkctl.c +++ b/src/varlinkctl/varlinkctl.c @@ -1269,7 +1269,7 @@ static int verb_serve(int argc, char *argv[], uintptr_t _data, void *userdata) { if (r < 0) return log_error_errno(r, "Failed to get event loop: %m"); - r = sd_varlink_server_new(&s, SD_VARLINK_SERVER_INHERIT_USERDATA); + r = sd_varlink_server_new(&s, SD_VARLINK_SERVER_INHERIT_USERDATA|SD_VARLINK_SERVER_UPGRADABLE); if (r < 0) return log_error_errno(r, "Failed to allocate varlink server: %m"); From cd6b57ff7060c344c7f3a2a779f2618e488bddaa Mon Sep 17 00:00:00 2001 From: Michael Vogt Date: Tue, 7 Apr 2026 17:54:28 +0200 Subject: [PATCH 6/6] sd-varlink: use MSG_PEEK for protocol_upgrade connections When there is a potential protocol upgrade we need to be careful that we do not read beyond our json message as the custom protocol may be anything. This was archived via a byte-by-byte read. This is of course very inefficient. So this commit moves to use MSG_PEEK to find the boundary of the json message instead. This makes the performance hit a lot smaller. Thanks to Lennart for suggesting this. --- src/libsystemd/sd-varlink/sd-varlink.c | 54 +++++++++++++++++++++++--- 1 file changed, 48 insertions(+), 6 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index bc7e93f407..fe2bf0e638 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -844,10 +844,49 @@ static int varlink_write(sd_varlink *v) { #define VARLINK_FDS_MAX (16U*1024U) +static bool varlink_may_protocol_upgrade(sd_varlink *v) { + return v->protocol_upgrade || (v->server && FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE)); +} + +/* When a protocol upgrade might happen, peek at the socket data to find the \0 message + * boundary and return a read size that won't consume past it. This prevents over-reading + * raw post-upgrade data into the varlink input buffer. Falls back to byte-by-byte for + * non-socket fds where MSG_PEEK is not available. */ +static ssize_t varlink_peek_upgrade_boundary(sd_varlink *v, void *p, size_t rs) { + assert(v); + + if (!varlink_may_protocol_upgrade(v)) + return rs; + + if (v->prefer_read) + return 1; + + ssize_t peeked = recv(v->input_fd, p, rs, MSG_PEEK|MSG_DONTWAIT); + if (peeked < 0) { + if (errno == ENOTSOCK) { + v->prefer_read = true; + return 1; /* Not a socket, fall back to byte-to-byte */ + } else if (!ERRNO_IS_TRANSIENT(errno)) + return -errno; + + /* Transient error, this should not happen but fall back to byte-to-byte */ + return 1; + } + /* EOF, the real recv() will also get it so what we return does not matter */ + if (peeked == 0) + return rs; + + void *nul_chr = memchr(p, 0, peeked); + if (nul_chr) + return (ssize_t) ((char*) nul_chr - (char*) p) + 1; + + return peeked; +} + static int varlink_read(sd_varlink *v) { struct iovec iov; struct msghdr mh; - size_t rs; + ssize_t rs; ssize_t n; void *p; @@ -895,11 +934,14 @@ static int varlink_read(sd_varlink *v) { p = 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); + 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. Use MSG_PEEK to find the \0 message boundary and only consume up to it. For non-socket + * fds (pipes) MSG_PEEK is not available, so fall back to byte-by-byte reading. */ + rs = varlink_peek_upgrade_boundary(v, p, rs); + if (rs < 0) + return varlink_log_errno(v, rs, "Failed to peek upgrade boundary: %m"); if (v->allow_fd_passing_input > 0) { iov = IOVEC_MAKE(p, rs);