From 755f1cff89acaa1b28667f2b7cc0d6f4ee47c44d Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Jul 2026 10:45:08 +0200 Subject: [PATCH 1/5] json-stream: add helper json_stream_has_buffered_output() --- src/libsystemd/sd-json/json-stream.c | 19 ++++++++++++++++--- src/libsystemd/sd-json/json-stream.h | 1 + 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/src/libsystemd/sd-json/json-stream.c b/src/libsystemd/sd-json/json-stream.c index 88473b8192..ed4b6ca5b1 100644 --- a/src/libsystemd/sd-json/json-stream.c +++ b/src/libsystemd/sd-json/json-stream.c @@ -844,16 +844,29 @@ fail: return r; } +int json_stream_has_buffered_output(const JsonStream *s) { + assert(s); + + if (s->output_buffer_size == 0 && !s->output_queue) + return false; + + if (FLAGS_SET(s->flags, JSON_STREAM_WRITE_DISCONNECTED)) + return -ECONNRESET; + + return true; +} + int json_stream_flush(JsonStream *s) { int ret = 0, r; assert(s); for (;;) { - if (s->output_buffer_size == 0 && !s->output_queue) + r = json_stream_has_buffered_output(s); + if (r < 0) + return r; + if (r == 0) break; - if (FLAGS_SET(s->flags, JSON_STREAM_WRITE_DISCONNECTED)) - return -ECONNRESET; r = json_stream_write(s); if (r < 0) diff --git a/src/libsystemd/sd-json/json-stream.h b/src/libsystemd/sd-json/json-stream.h index b502c98676..336ececc0f 100644 --- a/src/libsystemd/sd-json/json-stream.h +++ b/src/libsystemd/sd-json/json-stream.h @@ -206,6 +206,7 @@ int json_stream_parse(JsonStream *s, sd_json_variant **ret); /* Status accessors used by the consumer's state machine. */ bool json_stream_has_buffered_input(const JsonStream *s); +int json_stream_has_buffered_output(const JsonStream *s); /* Compute the poll events the consumer should wait for. The stream queries the consumer's * phase via the registered get_phase callback. In JSON_STREAM_PHASE_READING the stream asks From 260fcf97639cb47f89d7dc7dfad11d360662488a Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Jul 2026 10:59:40 +0200 Subject: [PATCH 2/5] sd-varlink: fold separate 'protocol_upgrade' flag into state machine --- src/libsystemd/sd-varlink/sd-varlink.c | 105 ++++++++++--------- src/libsystemd/sd-varlink/varlink-internal.h | 16 ++- 2 files changed, 62 insertions(+), 59 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 048dbb4747..ea616a6fd3 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -48,27 +48,29 @@ #define VARLINK_COLLECT_MAX 1024U static const char* const varlink_state_table[_VARLINK_STATE_MAX] = { - [VARLINK_IDLE_CLIENT] = "idle-client", - [VARLINK_AWAITING_REPLY] = "awaiting-reply", - [VARLINK_AWAITING_REPLY_MORE] = "awaiting-reply-more", - [VARLINK_CALLING] = "calling", - [VARLINK_CALLED] = "called", - [VARLINK_COLLECTING] = "collecting", - [VARLINK_COLLECTING_REPLY] = "collecting-reply", - [VARLINK_PROCESSING_REPLY] = "processing-reply", - [VARLINK_IDLE_SERVER] = "idle-server", - [VARLINK_PROCESSING_METHOD] = "processing-method", - [VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more", - [VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway", - [VARLINK_PROCESSED_METHOD] = "processed-method", - [VARLINK_PENDING_METHOD] = "pending-method", - [VARLINK_PENDING_METHOD_MORE] = "pending-method-more", - [VARLINK_PENDING_DISCONNECT] = "pending-disconnect", - [VARLINK_PENDING_TIMEOUT] = "pending-timeout", - [VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect", - [VARLINK_PROCESSING_TIMEOUT] = "processing-timeout", - [VARLINK_PROCESSING_FAILURE] = "processing-failure", - [VARLINK_DISCONNECTED] = "disconnected", + [VARLINK_IDLE_CLIENT] = "idle-client", + [VARLINK_AWAITING_REPLY] = "awaiting-reply", + [VARLINK_AWAITING_REPLY_MORE] = "awaiting-reply-more", + [VARLINK_CALLING] = "calling", + [VARLINK_CALLED] = "called", + [VARLINK_COLLECTING] = "collecting", + [VARLINK_COLLECTING_REPLY] = "collecting-reply", + [VARLINK_PROCESSING_REPLY] = "processing-reply", + [VARLINK_IDLE_SERVER] = "idle-server", + [VARLINK_PROCESSING_METHOD] = "processing-method", + [VARLINK_PROCESSING_METHOD_MORE] = "processing-method-more", + [VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway", + [VARLINK_PROCESSING_METHOD_UPGRADE] = "processing-method-upgrade", + [VARLINK_PROCESSED_METHOD] = "processed-method", + [VARLINK_PENDING_METHOD] = "pending-method", + [VARLINK_PENDING_METHOD_MORE] = "pending-method-more", + [VARLINK_PENDING_METHOD_UPGRADE] = "pending-method-upgrade", + [VARLINK_PENDING_DISCONNECT] = "pending-disconnect", + [VARLINK_PENDING_TIMEOUT] = "pending-timeout", + [VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect", + [VARLINK_PROCESSING_TIMEOUT] = "processing-timeout", + [VARLINK_PROCESSING_FAILURE] = "processing-failure", + [VARLINK_DISCONNECTED] = "disconnected", }; DEFINE_PRIVATE_STRING_TABLE_LOOKUP_TO_STRING(varlink_state, VarlinkState); @@ -112,7 +114,7 @@ static JsonStreamPhase varlink_phase(void *userdata) { if (v->state == VARLINK_IDLE_CLIENT) return JSON_STREAM_PHASE_IDLE_CLIENT; - if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) return JSON_STREAM_PHASE_PENDING_OUTPUT; return JSON_STREAM_PHASE_OTHER; @@ -1096,7 +1098,7 @@ static int varlink_fiber_entry(void *userdata) { * VARLINK_PROCESSING_METHOD{,_MORE} to VARLINK_PENDING_METHOD{,_MORE}, so that's what we match * here to decide whether the call still needs a reply. Any other state (e.g. IDLE_SERVER after * the callback replied, or DISCONNECTED after sd_varlink_close()) means no fixup is needed. */ - if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) return r; if (r < 0) { @@ -1118,7 +1120,7 @@ static int varlink_fiber_entry(void *userdata) { r = 0; /* If we didn't manage to enqueue a response, then fail the connection completely. */ - if (r < 0 && IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + if (r < 0 && IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) goto fail; return r; @@ -1264,20 +1266,20 @@ static int varlink_dispatch_method(sd_varlink *v) { if (r < 0) goto fail; - varlink_set_state(v, (flags & SD_VARLINK_METHOD_MORE) ? VARLINK_PROCESSING_METHOD_MORE : - (flags & SD_VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY : - VARLINK_PROCESSING_METHOD); + varlink_set_state(v, (flags & SD_VARLINK_METHOD_MORE) ? VARLINK_PROCESSING_METHOD_MORE : + (flags & SD_VARLINK_METHOD_ONEWAY) ? VARLINK_PROCESSING_METHOD_ONEWAY : + (flags & SD_VARLINK_METHOD_UPGRADE) ? VARLINK_PROCESSING_METHOD_UPGRADE : + VARLINK_PROCESSING_METHOD); assert(v->server); /* Reset the per-call upgrade marker on every dispatch — a previous method's * UPGRADE flag must not bleed into this one. The transport-level bounded reads * stay active for SD_VARLINK_SERVER_UPGRADABLE servers regardless. */ - v->protocol_upgrade = FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE); json_stream_set_flags( &v->stream, JSON_STREAM_BOUNDED_READS, - v->protocol_upgrade || FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE)); + FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE) || FLAGS_SET(v->server->flags, SD_VARLINK_SERVER_UPGRADABLE)); /* First consult user supplied method implementations */ bool is_fiber = false; @@ -1401,6 +1403,10 @@ static int varlink_dispatch_method(sd_varlink *v) { varlink_set_state(v, VARLINK_PENDING_METHOD_MORE); break; + case VARLINK_PROCESSING_METHOD_UPGRADE: /* Method call wasn't replied to, will be replied to later */ + varlink_set_state(v, VARLINK_PENDING_METHOD_UPGRADE); + break; + case VARLINK_DISCONNECTED: /* Handler called sd_varlink_close() on us, which is fine */ break; @@ -1516,7 +1522,7 @@ _public_ int sd_varlink_dispatch_again(sd_varlink *v) { if (v->state == VARLINK_DISCONNECTED) return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); - if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + if (!IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection has no pending method."); varlink_set_state(v, VARLINK_IDLE_SERVER); @@ -2076,11 +2082,9 @@ _public_ int sd_varlink_call_and_upgrade( if (r < 0) return varlink_log_errno(v, r, "Failed to build json message: %m"); - v->protocol_upgrade = true; json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, true); r = varlink_call_internal(v, m); if (r < 0) { - v->protocol_upgrade = false; json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false); return r; } @@ -2125,7 +2129,6 @@ _public_ int sd_varlink_call_and_upgrade( return 1; finish: - v->protocol_upgrade = false; json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, false); assert(v->n_pending == 1); v->n_pending--; @@ -2360,8 +2363,8 @@ static int varlink_reply_internal(sd_varlink *v, sd_json_variant *parameters, bo return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); if (!IN_SET(v->state, - VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, - VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_UPGRADE, + VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy."); bool more = IN_SET(v->state, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PENDING_METHOD_MORE); @@ -2413,7 +2416,7 @@ static int varlink_reply_internal(sd_varlink *v, sd_json_variant *parameters, bo if (r < 0) return varlink_log_errno(v, r, "Failed to enqueue json message: %m"); - if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) { + if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) { /* We just replied to a method call that was let hanging for a while (i.e. we were outside of * the varlink_dispatch_method() stack frame), which means with this reply we are ready to * process further messages. */ @@ -2463,21 +2466,23 @@ _public_ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parame if (v->state == VARLINK_DISCONNECTED) return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); + /* Return -EBUSY if we are not processing methods at all */ if (!IN_SET(v->state, - VARLINK_PROCESSING_METHOD, - VARLINK_PENDING_METHOD)) + VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_ONEWAY, VARLINK_PROCESSING_METHOD_UPGRADE, + VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) 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."); + /* Return -EPROTO if we are processing methods, but an upgrade was not requested. */ + if (!IN_SET(v->state, + VARLINK_PROCESSING_METHOD_UPGRADE, + VARLINK_PENDING_METHOD_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. */ + /* 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 disabled unbounded reads in dispatch to limit subsequent reads to single bytes. But a + * misbehaving client could pipeline data early. */ if (json_stream_has_buffered_input(&v->stream)) return varlink_log_errno(v, SYNTHETIC_ERRNO(EBADMSG), "Unexpected buffered data from client during protocol upgrade."); @@ -2548,8 +2553,8 @@ _public_ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_varia if (v->state == VARLINK_DISCONNECTED) return varlink_log_errno(v, SYNTHETIC_ERRNO(ENOTCONN), "Not connected."); if (!IN_SET(v->state, - VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, - VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) + VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_UPGRADE, + VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "Connection busy."); if (v->previous) { @@ -2601,7 +2606,7 @@ _public_ int sd_varlink_error(sd_varlink *v, const char *error_id, sd_json_varia if (r < 0) return varlink_log_errno(v, r, "Failed to enqueue json message: %m"); - if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE)) { + if (IN_SET(v->state, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE)) { varlink_clear_current(v); varlink_set_state(v, VARLINK_IDLE_SERVER); } else diff --git a/src/libsystemd/sd-varlink/varlink-internal.h b/src/libsystemd/sd-varlink/varlink-internal.h index beec5be42c..16723d34a6 100644 --- a/src/libsystemd/sd-varlink/varlink-internal.h +++ b/src/libsystemd/sd-varlink/varlink-internal.h @@ -26,9 +26,11 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD, VARLINK_PROCESSING_METHOD_MORE, VARLINK_PROCESSING_METHOD_ONEWAY, + VARLINK_PROCESSING_METHOD_UPGRADE, VARLINK_PROCESSED_METHOD, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, + VARLINK_PENDING_METHOD_UPGRADE, /* Common states (only during shutdown) */ VARLINK_PENDING_DISCONNECT, @@ -60,16 +62,19 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD, \ VARLINK_PROCESSING_METHOD_MORE, \ VARLINK_PROCESSING_METHOD_ONEWAY, \ + VARLINK_PROCESSING_METHOD_UPGRADE, \ VARLINK_PROCESSED_METHOD, \ VARLINK_PENDING_METHOD, \ - VARLINK_PENDING_METHOD_MORE) + VARLINK_PENDING_METHOD_MORE, \ + VARLINK_PENDING_METHOD_UPGRADE) /* Tests whether we are expected to generate a method call reply, i.e. are processing a method call, except * one with the ONEWAY flag set. */ #define VARLINK_STATE_WANTS_REPLY(state) \ IN_SET(state, \ VARLINK_PROCESSING_METHOD, \ - VARLINK_PROCESSING_METHOD_MORE) + VARLINK_PROCESSING_METHOD_MORE, \ + VARLINK_PROCESSING_METHOD_UPGRADE) typedef struct sd_varlink { unsigned n_ref; @@ -86,13 +91,6 @@ typedef struct sd_varlink { unsigned n_pending; - /* Per-call protocol-upgrade marker: set when the *current* method call carries the - * SD_VARLINK_METHOD_UPGRADE flag. Validated by sd_varlink_reply_and_upgrade() to - * ensure the caller's contract is honored. The transport-layer "stop reading at the - * next message boundary" behavior is governed independently by the JsonStream's - * bounded_reads flag. */ - bool protocol_upgrade; - sd_varlink_reply_t reply_callback; sd_json_variant *current; From f194afa164ad5d1eeec0cb4d7a426dce48dad6b7 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Jul 2026 11:13:38 +0200 Subject: [PATCH 3/5] sd-varlink: move varlink_handle_upgrade_fds() before sd_varlink_process() This is a simple move, in preparation for using this as part of the sd_varlink_process() callchain. --- src/libsystemd/sd-varlink/sd-varlink.c | 94 +++++++++++++------------- 1 file changed, 47 insertions(+), 47 deletions(-) diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index ea616a6fd3..dd1e01c5ba 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -1427,6 +1427,53 @@ fail: return r; } +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 (json_stream_has_buffered_input(&v->stream)) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO), + "Unexpected buffered data during protocol upgrade, refusing."); + + _cleanup_close_ int input_fd = TAKE_FD(v->stream.input_fd), + output_fd = TAKE_FD(v->stream.output_fd); + + /* 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. For bidirectional sockets (input_fd == output_fd), dup the fd so that + * callers always get two independent fds they can close separately. */ + if (input_fd == output_fd) { + output_fd = fcntl(input_fd, F_DUPFD_CLOEXEC, 3); + if (output_fd < 0) + return varlink_log_errno(v, errno, "Failed to dup upgraded connection fd: %m"); + } else { + r = fd_nonblock(output_fd, false); + if (r < 0) + return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m"); + } + + r = fd_nonblock(input_fd, false); + if (r < 0) + return varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m"); + + /* Hand out requested fds, shut down unwanted directions. */ + if (ret_input_fd) + *ret_input_fd = TAKE_FD(input_fd); + else + (void) shutdown(input_fd, SHUT_RD); + + if (ret_output_fd) + *ret_output_fd = TAKE_FD(output_fd); + else + (void) shutdown(output_fd, SHUT_WR); + + return 0; +} + _public_ int sd_varlink_process(sd_varlink *v) { int r; @@ -2011,53 +2058,6 @@ _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 (json_stream_has_buffered_input(&v->stream)) - return varlink_log_errno(v, SYNTHETIC_ERRNO(EPROTO), - "Unexpected buffered data during protocol upgrade, refusing."); - - _cleanup_close_ int input_fd = TAKE_FD(v->stream.input_fd), - output_fd = TAKE_FD(v->stream.output_fd); - - /* 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. For bidirectional sockets (input_fd == output_fd), dup the fd so that - * callers always get two independent fds they can close separately. */ - if (input_fd == output_fd) { - output_fd = fcntl(input_fd, F_DUPFD_CLOEXEC, 3); - if (output_fd < 0) - return varlink_log_errno(v, errno, "Failed to dup upgraded connection fd: %m"); - } else { - r = fd_nonblock(output_fd, false); - if (r < 0) - return varlink_log_errno(v, r, "Failed to set output fd to blocking mode: %m"); - } - - r = fd_nonblock(input_fd, false); - if (r < 0) - return varlink_log_errno(v, r, "Failed to set input fd to blocking mode: %m"); - - /* Hand out requested fds, shut down unwanted directions. */ - if (ret_input_fd) - *ret_input_fd = TAKE_FD(input_fd); - else - (void) shutdown(input_fd, SHUT_RD); - - if (ret_output_fd) - *ret_output_fd = TAKE_FD(output_fd); - else - (void) shutdown(output_fd, SHUT_WR); - - return 0; -} - _public_ int sd_varlink_call_and_upgrade( sd_varlink *v, const char *method, From 746d10d106ab367316f2965dd6b0e7b5a40e8e52 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Jul 2026 10:47:46 +0200 Subject: [PATCH 4/5] sd-varlink: add async server-side upgrade API --- man/rules/meson.build | 8 + man/sd_varlink_reply_and_upgrade.xml | 227 +++++++++++++++++++ src/libsystemd/libsystemd.sym | 7 + src/libsystemd/sd-varlink/sd-varlink.c | 113 ++++++++- src/libsystemd/sd-varlink/test-varlink.c | 121 ++++++++++ src/libsystemd/sd-varlink/varlink-internal.h | 7 +- src/systemd/sd-varlink.h | 10 + 7 files changed, 490 insertions(+), 3 deletions(-) create mode 100644 man/sd_varlink_reply_and_upgrade.xml diff --git a/man/rules/meson.build b/man/rules/meson.build index 22e4384251..8c21e90090 100644 --- a/man/rules/meson.build +++ b/man/rules/meson.build @@ -977,6 +977,14 @@ manpages = [ ''], ['sd_varlink_is_connected', '3', ['sd_varlink_is_idle'], ''], ['sd_varlink_push_fd', '3', ['sd_varlink_push_dup_fd'], ''], + ['sd_varlink_reply_and_upgrade', + '3', + ['sd_varlink_bind_upgrade', + 'sd_varlink_respond_and_upgrade', + 'sd_varlink_respond_and_upgradeb', + 'sd_varlink_respond_and_upgradebo', + 'sd_varlink_upgrade_t'], + ''], ['sd_varlink_send', '3', ['sd_varlink_sendb', 'sd_varlink_sendbo'], ''], ['sd_varlink_server_listen_address', '3', diff --git a/man/sd_varlink_reply_and_upgrade.xml b/man/sd_varlink_reply_and_upgrade.xml new file mode 100644 index 0000000000..f51dee3779 --- /dev/null +++ b/man/sd_varlink_reply_and_upgrade.xml @@ -0,0 +1,227 @@ + + + + + + + + sd_varlink_reply_and_upgrade + systemd + + + + sd_varlink_reply_and_upgrade + 3 + + + + sd_varlink_reply_and_upgrade + sd_varlink_respond_and_upgrade + sd_varlink_respond_and_upgradeb + sd_varlink_respond_and_upgradebo + sd_varlink_bind_upgrade + sd_varlink_upgrade_t + + Reply to a Varlink method call and take over the connection for a raw protocol + + + + + #include <systemd/sd-varlink.h> + + typedef int (*sd_varlink_upgrade_t)(sd_varlink *link, int input_fd, int output_fd, void *userdata); + + + int sd_varlink_reply_and_upgrade + sd_varlink *link + sd_json_variant *parameters + int *ret_input_fd + int *ret_output_fd + + + + int sd_varlink_respond_and_upgrade + sd_varlink *link + sd_json_variant *parameters + + + + int sd_varlink_respond_and_upgradeb + sd_varlink *link + … + + + + int sd_varlink_respond_and_upgradebo + sd_varlink *link + … + + + + int sd_varlink_bind_upgrade + sd_varlink *link + sd_varlink_upgrade_t callback + + + + + + Description + + These functions implement the server side of a Varlink protocol upgrade. A + Varlink client may request an upgrade when invoking a method (see + sd_varlink_call3 and + the SD_VARLINK_METHOD_UPGRADE flag passed to method callbacks). If a method is + invoked this way, the server may send one final reply and then abandon the Varlink protocol on the + connection entirely, handing the underlying socket or pipe file descriptors back to the caller so that an + arbitrary, non-Varlink ("raw") protocol may be spoken over them instead. This is useful for methods that + negotiate a switch to a streaming or binary protocol, in a manner similar to the + Upgrade: mechanism of HTTP. + + sd_varlink_reply_and_upgrade() sends a final reply to the currently processed + method call and then takes over the connection's file descriptors synchronously. It + takes the Varlink connection object, a JSON object with the reply parameters (which may be + NULL), and two output pointers for the file descriptors. The parameters + object is validated against the method's declared reply signature, if known. The reply is flushed to the + connection before the file descriptors are returned, so that the client has received the full reply before + the caller starts speaking the upgraded protocol. On success the connection is detached from its event + loop (if any) and disconnected, and ownership of the file descriptors is transferred to the caller, which + is responsible for eventually closing them with + close2. + + The returned file descriptors are switched to blocking mode. For bidirectional sockets a single + underlying file descriptor carries both directions; in this case ret_input_fd and + ret_output_fd are returned as two independent (duplicated) descriptors referring + to the same socket, so they may be closed separately. For transports backed by a pair of pipes the two + descriptors differ and refer to the distinct read and write ends. Either + ret_input_fd or ret_output_fd (but not both) may be + NULL; in that case the unwanted direction is shut down (if they refer to the same + socket, see shutdown2) or + closed (if separate file descriptors) rather than returned. + + Note that sd_varlink_reply_and_upgrade() blocks synchronously until the reply + has been flushed to the socket. A misbehaving or adversarial client that stops reading could stall the + caller. It should hence not be used in servers that multiplex many Varlink connections in a single event + loop, as one stalled connection would block progress on all others. Use + sd_varlink_respond_and_upgrade() in that case instead. + + sd_varlink_respond_and_upgrade() is the asynchronous + counterpart of sd_varlink_reply_and_upgrade(). It enqueues the final reply and + returns immediately, without blocking and without returning the file descriptors directly. The reply is + subsequently flushed by the event loop during regular + sd_varlink_process3 + invocations. Once the reply has been fully flushed, the connection's file descriptors are handed to the + upgrade callback previously registered with sd_varlink_bind_upgrade(), and the + Varlink connection is disconnected. This variant is the right choice for servers that handle many + connections concurrently, as it never blocks on a single client. + + sd_varlink_respond_and_upgradeb() is similar to + sd_varlink_respond_and_upgrade(), but instead of expecting a fully constructed + sd_json_variant object carrying the reply parameters, this object is constructed on-the-fly + from the variadic argument list, in a style identical to + sd_json_build3. + sd_varlink_respond_and_upgradebo() is identical to + sd_varlink_respond_and_upgradeb(), but an enclosing JSON object is added implicitly, + so that the argument list is expected to consist of object field pairs only, in a style identical to + sd_json_buildo3. + + sd_varlink_bind_upgrade() registers the upgrade + callback that is invoked once a reply enqueued via + sd_varlink_respond_and_upgrade() (or one of its variants) has been flushed to the + connection. The callback is of type sd_varlink_upgrade_t and receives the Varlink connection + object, the input and output file descriptors of the (now upgraded) connection, and the userdata pointer + associated with the connection (see + sd_varlink_set_userdata3). + Ownership of the two file descriptors is transferred to the callback (regardless if the callback later + fails or not). The descriptor semantics (blocking mode, duplication for bidirectional sockets, distinct + ends for pipe pairs) match those described for sd_varlink_reply_and_upgrade() + above. Passing NULL as callback removes a previously + registered callback. + + + + Return Value + + On success, sd_varlink_reply_and_upgrade(), + sd_varlink_respond_and_upgrade(), + sd_varlink_respond_and_upgradeb() and + sd_varlink_respond_and_upgradebo() return a positive integer. + sd_varlink_bind_upgrade() returns a non-negative integer on success. On failure, they + all return a negative errno-style error code. + + + Errors + + Returned errors may indicate the following problems: + + + + -EINVAL + + An argument is invalid. + + + + -ENOTCONN + + The Varlink connection object is not connected. + + + + -EBUSY + + The connection is not currently processing a method call, so there is no reply to + send. When returned by sd_varlink_bind_upgrade(), a different upgrade callback + was already registered on the connection. + + + + -EPROTO + + The currently processed method call did not request a protocol upgrade, hence the + connection may not be taken over. + + + + -EBADMSG + + The client sent unexpected data before the upgrade completed, i.e. it pipelined + raw protocol data before receiving the upgrade reply. + + + + -ENOMEM + + Memory allocation failed. + + + + + + + + + History + sd_varlink_reply_and_upgrade() was added in version 261. + sd_varlink_respond_and_upgrade(), + sd_varlink_respond_and_upgradeb(), + sd_varlink_respond_and_upgradebo() and + sd_varlink_bind_upgrade() were added in version 262. + + + + See Also + + + systemd1 + sd-varlink3 + sd_varlink_reply3 + sd_varlink_call3 + sd_varlink_process3 + sd_varlink_set_userdata3 + sd_json_build3 + + + diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym index 6afc11d91b..317c51b4c3 100644 --- a/src/libsystemd/libsystemd.sym +++ b/src/libsystemd/libsystemd.sym @@ -1104,3 +1104,10 @@ global: sd_varlink_reply_and_upgrade; sd_varlink_set_sentinel; } LIBSYSTEMD_260; + +LIBSYSTEMD_262 { +global: + sd_varlink_bind_upgrade; + sd_varlink_respond_and_upgrade; + sd_varlink_respond_and_upgradeb; +} LIBSYSTEMD_261; diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index dd1e01c5ba..198883d754 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -62,9 +62,11 @@ static const char* const varlink_state_table[_VARLINK_STATE_MAX] = { [VARLINK_PROCESSING_METHOD_ONEWAY] = "processing-method-oneway", [VARLINK_PROCESSING_METHOD_UPGRADE] = "processing-method-upgrade", [VARLINK_PROCESSED_METHOD] = "processed-method", + [VARLINK_PROCESSED_METHOD_UPGRADE] = "processed-method-upgrade", [VARLINK_PENDING_METHOD] = "pending-method", [VARLINK_PENDING_METHOD_MORE] = "pending-method-more", [VARLINK_PENDING_METHOD_UPGRADE] = "pending-method-upgrade", + [VARLINK_UPGRADING] = "upgrading", [VARLINK_PENDING_DISCONNECT] = "pending-disconnect", [VARLINK_PENDING_TIMEOUT] = "pending-timeout", [VARLINK_PROCESSING_DISCONNECT] = "processing-disconnect", @@ -1407,6 +1409,11 @@ static int varlink_dispatch_method(sd_varlink *v) { varlink_set_state(v, VARLINK_PENDING_METHOD_UPGRADE); break; + case VARLINK_PROCESSED_METHOD_UPGRADE: /* Upgrade requested, let's now wait till everything is flushed */ + varlink_clear_current(v); + varlink_set_state(v, VARLINK_UPGRADING); + break; + case VARLINK_DISCONNECTED: /* Handler called sd_varlink_close() on us, which is fine */ break; @@ -1474,6 +1481,41 @@ static int varlink_handle_upgrade_fds(sd_varlink *v, int *ret_input_fd, int *ret return 0; } +static int varlink_dispatch_upgrade(sd_varlink *v) { + int r; + + assert(v); + + if (v->state != VARLINK_UPGRADING) + return 0; + + r = json_stream_has_buffered_output(&v->stream); + if (r < 0) + return r; + if (r > 0) + return 0; + + /* Detach from the event loop before stealing the fds */ + sd_varlink_detach_event(v); + + _cleanup_close_ int input_fd = -EBADF, output_fd = -EBADF; + r = varlink_handle_upgrade_fds(v, &input_fd, &output_fd); + sd_varlink_close(v); + if (r < 0) + return r; + + /* Fail if by this time the caller still hasn't set an upgrade callback. */ + if (!v->upgrade_callback) + return -ENOPKG; + + /* No matter what, we donate the fds to callback */ + r = v->upgrade_callback(v, TAKE_FD(input_fd), TAKE_FD(output_fd), v->userdata); + if (r < 0) + return r; + + return 1; +} + _public_ int sd_varlink_process(sd_varlink *v) { int r; @@ -1490,6 +1532,12 @@ _public_ int sd_varlink_process(sd_varlink *v) { if (r != 0) goto finish; + r = varlink_dispatch_upgrade(v); + if (r < 0) + varlink_log_errno(v, r, "Upgrade dispatch failed: %m"); + if (r != 0) + goto finish; + r = varlink_dispatch_reply(v); if (r < 0) varlink_log_errno(v, r, "Reply dispatch failed: %m"); @@ -2457,11 +2505,12 @@ _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) { +_public_ int sd_varlink_respond_and_upgrade( + sd_varlink *v, + sd_json_variant *parameters) { 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."); @@ -2508,6 +2557,55 @@ _public_ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parame if (r < 0) return varlink_log_errno(v, r, "Failed to enqueue json message: %m"); + switch (v->state) { + + case VARLINK_PENDING_METHOD_UPGRADE: + varlink_clear_current(v); + varlink_set_state(v, VARLINK_UPGRADING); + break; + + case VARLINK_PROCESSING_METHOD_UPGRADE: + varlink_set_state(v, VARLINK_PROCESSED_METHOD_UPGRADE); + break; + + default: + assert_not_reached(); + } + + return 1; +} + +_public_ int sd_varlink_respond_and_upgradeb(sd_varlink *v, ...) { + _cleanup_(sd_json_variant_unrefp) sd_json_variant *parameters = NULL; + va_list ap; + int r; + + assert_return(v, -EINVAL); + + va_start(ap, v); + r = sd_json_buildv(¶meters, ap); + va_end(ap); + + if (r < 0) + return r; + + return sd_varlink_respond_and_upgrade(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); + + r = sd_varlink_respond_and_upgrade(v, parameters); + if (r < 0) + return r; + /* Flush the reply to the socket before stealing the fds. The reply must be fully written * before the caller starts speaking the upgraded protocol. */ r = json_stream_flush(&v->stream); @@ -2797,6 +2895,17 @@ _public_ int sd_varlink_bind_reply(sd_varlink *v, sd_varlink_reply_t reply) { return 0; } +_public_ int sd_varlink_bind_upgrade(sd_varlink *v, sd_varlink_upgrade_t upgrade) { + assert_return(v, -EINVAL); + + if (upgrade && v->upgrade_callback && upgrade != v->upgrade_callback) + return varlink_log_errno(v, SYNTHETIC_ERRNO(EBUSY), "A different callback was already set."); + + v->upgrade_callback = upgrade; + + return 0; +} + _public_ void* sd_varlink_set_userdata(sd_varlink *v, void *userdata) { void *old; diff --git a/src/libsystemd/sd-varlink/test-varlink.c b/src/libsystemd/sd-varlink/test-varlink.c index a2a32b7f78..73b63c45d7 100644 --- a/src/libsystemd/sd-varlink/test-varlink.c +++ b/src/libsystemd/sd-varlink/test-varlink.c @@ -1177,6 +1177,127 @@ TEST(upgrade_pipelining) { ASSERT_OK(sd_future_result(f)); } +/* Stashes the fds handed to the asynchronous upgrade callback so the raw-I/O fiber can pick them up. + * A single connection is upgraded per test run, so globals are sufficient. */ +static int respond_upgrade_input_fd = -EBADF; +static int respond_upgrade_output_fd = -EBADF; + +static int respond_upgrade_io_fiber(void *arg) { + _cleanup_close_ int input_fd = TAKE_FD(respond_upgrade_input_fd), output_fd = TAKE_FD(respond_upgrade_output_fd); + + /* Same reverse-echo raw protocol as method_upgrade(), but reached through the asynchronous + * upgrade callback instead of a synchronous fiber method: read until the client shuts down its + * write side, reverse the bytes, and write them back. */ + char buf[64] = {}; + ssize_t n = ASSERT_OK(loop_read(input_fd, buf, sizeof(buf) - 1, /* do_poll= */ true)); + ASSERT_GT(n, 0); + + for (ssize_t i = 0; i < n / 2; i++) + SWAP_TWO(buf[i], buf[n - 1 - i]); + + ASSERT_OK(loop_write(output_fd, buf, n)); + + return 0; +} + +static int respond_upgrade_callback(sd_varlink *link, int input_fd, int output_fd, void *userdata) { + sd_event *e = ASSERT_PTR(userdata); + + /* The upgrade callback is invoked from sd_varlink_process() in the main event loop (not a + * fiber), so it must not block on raw I/O. Hand the connection's fds to a floating fiber that + * speaks the raw protocol, allowing the client fiber to make progress concurrently. Ownership of + * the fds is transferred to us regardless of what happens next, so stash them before spawning. */ + respond_upgrade_input_fd = input_fd; + respond_upgrade_output_fd = output_fd; + + ASSERT_OK(sd_fiber_new(e, "respond-upgrade-io", respond_upgrade_io_fiber, /* userdata= */ NULL, /* destroy= */ NULL, /* ret= */ NULL)); + + return 0; +} + +static int method_respond_upgrade(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { + int r; + + ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_UPGRADE)); + + /* Register the async upgrade callback, then enqueue the final reply. Unlike + * sd_varlink_reply_and_upgrade(), sd_varlink_respond_and_upgradebo() returns immediately without + * blocking: the reply is flushed by the event loop and the fds are subsequently handed to + * respond_upgrade_callback(). This is why the method may be bound as a regular (non-fiber) method. */ + ASSERT_OK(sd_varlink_bind_upgrade(link, respond_upgrade_callback)); + + r = sd_varlink_respond_and_upgradebo( + link, + SD_JSON_BUILD_PAIR_STRING("greeting", "aloha")); + if (r < 0) + return r; + + return 0; +} + +static int respond_upgrade_client_fiber(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, "respond-upgrade-client")); + + ASSERT_OK(sd_varlink_call_and_upgrade(c, "io.test.RespondUpgrade", /* parameters= */ NULL, &o, &error_id, &input_fd, &output_fd)); + ASSERT_NULL(error_id); + + /* The reply parameters enqueued asynchronously by the server must have arrived intact. */ + ASSERT_STREQ(sd_json_variant_string(sd_json_variant_by_key(o, "greeting")), "aloha"); + + ASSERT_GE(input_fd, 0); + ASSERT_GE(output_fd, 0); + ASSERT_NE(input_fd, output_fd); /* library dups for bidirectional sockets */ + + /* Send a test string, shut down write side so the server sees EOF, then read the reversed reply */ + static const char msg[] = "Hello!"; + 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)); + ASSERT_EQ((size_t) n, strlen(msg)); + ASSERT_STREQ(buf, "!olleH"); + + ASSERT_OK(sd_event_exit(sd_fiber_get_event(), EXIT_SUCCESS)); + return 0; +} + +TEST(respond_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; + _cleanup_(sd_future_unrefp) sd_future *f = NULL; + 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, "respond-upgrade-server")); + /* The event loop is propagated to the async upgrade callback via the connection userdata (inherited + * from the server) so the callback can spawn the raw-I/O fiber. */ + sd_varlink_server_set_userdata(s, e); + /* Unlike TEST(upgrade), this is a *regular* (non-fiber) method: it drives the asynchronous + * sd_varlink_respond_and_upgradebo() API, which never blocks the event loop on a single client. */ + ASSERT_OK(sd_varlink_server_bind_method(s, "io.test.RespondUpgrade", method_respond_upgrade)); + ASSERT_OK(sd_varlink_server_listen_address(s, sp, 0600)); + ASSERT_OK(sd_varlink_server_attach_event(s, e, 0)); + + ASSERT_OK(sd_fiber_new(e, "respond-upgrade-client", respond_upgrade_client_fiber, (void*) sp, /* destroy= */ NULL, &f)); + + ASSERT_OK(sd_event_loop(e)); + + ASSERT_OK(sd_future_result(f)); +} + typedef struct ExecDirServer { sd_varlink_server *server; const char *name; diff --git a/src/libsystemd/sd-varlink/varlink-internal.h b/src/libsystemd/sd-varlink/varlink-internal.h index 16723d34a6..cefef889be 100644 --- a/src/libsystemd/sd-varlink/varlink-internal.h +++ b/src/libsystemd/sd-varlink/varlink-internal.h @@ -28,9 +28,11 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD_ONEWAY, VARLINK_PROCESSING_METHOD_UPGRADE, VARLINK_PROCESSED_METHOD, + VARLINK_PROCESSED_METHOD_UPGRADE, VARLINK_PENDING_METHOD, VARLINK_PENDING_METHOD_MORE, VARLINK_PENDING_METHOD_UPGRADE, + VARLINK_UPGRADING, /* Common states (only during shutdown) */ VARLINK_PENDING_DISCONNECT, @@ -64,9 +66,11 @@ typedef enum VarlinkState { VARLINK_PROCESSING_METHOD_ONEWAY, \ VARLINK_PROCESSING_METHOD_UPGRADE, \ VARLINK_PROCESSED_METHOD, \ + VARLINK_PROCESSED_METHOD_UPGRADE, \ VARLINK_PENDING_METHOD, \ VARLINK_PENDING_METHOD_MORE, \ - VARLINK_PENDING_METHOD_UPGRADE) + VARLINK_PENDING_METHOD_UPGRADE, \ + VARLINK_UPGRADING) /* Tests whether we are expected to generate a method call reply, i.e. are processing a method call, except * one with the ONEWAY flag set. */ @@ -92,6 +96,7 @@ typedef struct sd_varlink { unsigned n_pending; sd_varlink_reply_t reply_callback; + sd_varlink_upgrade_t upgrade_callback; sd_json_variant *current; sd_json_variant *current_collected; diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index 3be82a7ddb..7480594d74 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -84,6 +84,7 @@ __extension__ typedef enum _SD_ENUM_TYPE_S64(sd_varlink_invocation_flags_t) { typedef int (*sd_varlink_method_t)(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata); typedef int (*sd_varlink_reply_t)(sd_varlink *link, sd_json_variant *parameters, const char *error_id, sd_varlink_reply_flags_t flags, void *userdata); +typedef int (*sd_varlink_upgrade_t)(sd_varlink *link, int input_fd, int output_fd, void *userdata); typedef int (*sd_varlink_connect_t)(sd_varlink_server *server, sd_varlink *link, void *userdata); typedef void (*sd_varlink_disconnect_t)(sd_varlink_server *server, sd_varlink *link, void *userdata); @@ -182,6 +183,12 @@ int sd_varlink_replyb(sd_varlink *v, ...); * connections. */ int sd_varlink_reply_and_upgrade(sd_varlink *v, sd_json_variant *parameters, int *ret_input_fd, int *ret_output_fd); +/* Like sd_varlink_reply_and_upgrade() but asynchronous */ +int sd_varlink_respond_and_upgrade(sd_varlink *v, sd_json_variant *parameters); +int sd_varlink_respond_and_upgradeb(sd_varlink *v, ...); +#define sd_varlink_respond_and_upgradebo(v, ...) \ + sd_varlink_respond_and_upgradeb((v), SD_JSON_BUILD_OBJECT(__VA_ARGS__)) + /* 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, ...); @@ -224,6 +231,9 @@ int sd_varlink_set_allow_fd_passing_output(sd_varlink *v, int b); /* Bind a disconnect, reply or timeout callback */ int sd_varlink_bind_reply(sd_varlink *v, sd_varlink_reply_t reply); +/* Bind an upgrade callback (server side) */ +int sd_varlink_bind_upgrade(sd_varlink *v, sd_varlink_upgrade_t upgrade); + void* sd_varlink_set_userdata(sd_varlink *v, void *userdata); void* sd_varlink_get_userdata(sd_varlink *v); From 953d21846aad072103b5da00e2f67f7378335bb5 Mon Sep 17 00:00:00 2001 From: Lennart Poettering Date: Tue, 7 Jul 2026 12:45:31 +0200 Subject: [PATCH 5/5] sd-varlink: add sd_varlink_call_and_upgradeb() + sd_varlink_call_and_upgradebo() This are to the existing sd_varlink_call_and_upgrade() what sd_varlink_callb() and sd_varlink_callbo() are to sd_varlink_call(): they put together an object on the fly, via the usual JSON builder logic. --- man/rules/meson.build | 4 + man/sd_varlink_call_and_upgrade.xml | 177 +++++++++++++++++++++++++ man/sd_varlink_reply_and_upgrade.xml | 1 + src/libsystemd/libsystemd.sym | 1 + src/libsystemd/sd-varlink/sd-varlink.c | 25 ++++ src/systemd/sd-varlink.h | 3 + 6 files changed, 211 insertions(+) create mode 100644 man/sd_varlink_call_and_upgrade.xml diff --git a/man/rules/meson.build b/man/rules/meson.build index 8c21e90090..486f565815 100644 --- a/man/rules/meson.build +++ b/man/rules/meson.build @@ -968,6 +968,10 @@ manpages = [ 'sd_uid_get_sessions', 'sd_uid_is_on_seat'], 'HAVE_PAM'], + ['sd_varlink_call_and_upgrade', + '3', + ['sd_varlink_call_and_upgradeb', 'sd_varlink_call_and_upgradebo'], + ''], ['sd_varlink_connect_address', '3', ['sd_varlink_connect_exec', diff --git a/man/sd_varlink_call_and_upgrade.xml b/man/sd_varlink_call_and_upgrade.xml new file mode 100644 index 0000000000..48eb9de93f --- /dev/null +++ b/man/sd_varlink_call_and_upgrade.xml @@ -0,0 +1,177 @@ + + + + + + + + sd_varlink_call_and_upgrade + systemd + + + + sd_varlink_call_and_upgrade + 3 + + + + sd_varlink_call_and_upgrade + sd_varlink_call_and_upgradeb + sd_varlink_call_and_upgradebo + + Invoke a Varlink method and take over the connection for a raw protocol + + + + + #include <systemd/sd-varlink.h> + + + int sd_varlink_call_and_upgrade + sd_varlink *link + 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 + + + + int sd_varlink_call_and_upgradeb + sd_varlink *link + const char *method + sd_json_variant **ret_parameters + const char **ret_error_id + int *ret_input_fd + int *ret_output_fd + … + + + + int sd_varlink_call_and_upgradebo + sd_varlink *link + const char *method + sd_json_variant **ret_parameters + const char **ret_error_id + int *ret_input_fd + int *ret_output_fd + … + + + + + + Description + + These functions implement the client side of a Varlink protocol upgrade. They + are the counterpart of the server-side + sd_varlink_reply_and_upgrade3 + family. If a method call is issued this way, the connection may switch, after the reply, from the Varlink + protocol to an arbitrary, non-Varlink ("raw") protocol spoken over the underlying socket or pipe file + descriptors, in a manner similar to the Upgrade: mechanism of HTTP. + + sd_varlink_call_and_upgrade() synchronously invokes the method + method on the connection link, requesting a protocol upgrade, + and waits for the reply. It takes the Varlink connection object, the method name, a JSON object with the + method call parameters (which may be NULL), two output pointers for the reply + parameters and a possible error identifier, and two output pointers for the file descriptors. The reply + parameters returned in ret_parameters and the error identifier returned in + ret_error_id are borrowed references that remain valid only until the connection is + closed or unreffed. Either of the two may be passed as NULL if the information is not + needed. + + If the server replies with a regular reply (rather than a Varlink error) the connection is upgraded: + it is disconnected from the Varlink protocol, and ownership of the underlying file descriptors is + transferred to the caller via ret_input_fd and ret_output_fd, + which is responsible for eventually closing them with + close2. + If the server replies with a Varlink error instead, the connection is not upgraded and + no file descriptors are returned; the error identifier is reported in ret_error_id + (if non-NULL), otherwise a negative errno-style error derived from it is returned. + + The returned file descriptors are switched to blocking mode. For bidirectional sockets a single + underlying file descriptor carries both directions; in this case ret_input_fd and + ret_output_fd are returned as two independent (duplicated) descriptors referring to + the same socket, so they may be closed separately. For transports backed by a pair of pipes the two + descriptors differ and refer to the distinct read and write ends. At least one of + ret_input_fd or ret_output_fd must be + non-NULL. + + sd_varlink_call_and_upgradeb() is similar to + sd_varlink_call_and_upgrade(), but instead of expecting a fully constructed + sd_json_variant object carrying the method call parameters, this object is constructed + on-the-fly from the variadic argument list, in a style identical to + sd_json_build3. + sd_varlink_call_and_upgradebo() is identical to + sd_varlink_call_and_upgradeb(), but an enclosing JSON object is added implicitly, so + that the argument list is expected to consist of object field pairs only, in a style identical to + sd_json_buildo3. This + relationship mirrors that of + sd_varlink_call3 and its + sd_varlink_callb()/sd_varlink_callbo() variants. + + + + Return Value + + On success, these functions return a positive integer if the connection was upgraded, or zero if the + server returned a Varlink error and ret_error_id was set to it. On failure, they + return a negative errno-style error code. + + + Errors + + Returned errors may indicate the following problems: + + + + -EINVAL + + An argument is invalid. + + + + -ENOTCONN + + The Varlink connection object is not connected. + + + + -EPROTO + + A protocol error occurred, for example the server sent unexpected raw protocol data + before the upgrade completed. + + + + -ENOMEM + + Memory allocation failed. + + + + + + + + + History + sd_varlink_call_and_upgrade() was added in version 261. + sd_varlink_call_and_upgradeb() and + sd_varlink_call_and_upgradebo() were added in version 262. + + + + See Also + + + systemd1 + sd-varlink3 + sd_varlink_call3 + sd_varlink_reply_and_upgrade3 + sd_json_build3 + + + diff --git a/man/sd_varlink_reply_and_upgrade.xml b/man/sd_varlink_reply_and_upgrade.xml index f51dee3779..b2a8f6752f 100644 --- a/man/sd_varlink_reply_and_upgrade.xml +++ b/man/sd_varlink_reply_and_upgrade.xml @@ -219,6 +219,7 @@ sd-varlink3 sd_varlink_reply3 sd_varlink_call3 + sd_varlink_call_and_upgrade3 sd_varlink_process3 sd_varlink_set_userdata3 sd_json_build3 diff --git a/src/libsystemd/libsystemd.sym b/src/libsystemd/libsystemd.sym index 317c51b4c3..83fb8c92b3 100644 --- a/src/libsystemd/libsystemd.sym +++ b/src/libsystemd/libsystemd.sym @@ -1108,6 +1108,7 @@ global: LIBSYSTEMD_262 { global: sd_varlink_bind_upgrade; + sd_varlink_call_and_upgradeb; sd_varlink_respond_and_upgrade; sd_varlink_respond_and_upgradeb; } LIBSYSTEMD_261; diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index 198883d754..a86e523a47 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -2183,6 +2183,31 @@ finish: return r; } +_public_ int sd_varlink_call_and_upgradeb( + sd_varlink *v, + const char *method, + 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 *parameters = NULL; + va_list ap; + int r; + + assert_return(v, -EINVAL); + assert_return(method, -EINVAL); + + va_start(ap, ret_output_fd); + r = sd_json_buildv(¶meters, ap); + va_end(ap); + if (r < 0) + return varlink_log_errno(v, r, "Failed to build json message: %m"); + + return sd_varlink_call_and_upgrade(v, method, parameters, ret_parameters, ret_error_id, ret_input_fd, ret_output_fd); +} + _public_ int sd_varlink_callb_ap( sd_varlink *v, const char *method, diff --git a/src/systemd/sd-varlink.h b/src/systemd/sd-varlink.h index 7480594d74..3dbb443fa3 100644 --- a/src/systemd/sd-varlink.h +++ b/src/systemd/sd-varlink.h @@ -145,6 +145,9 @@ int sd_varlink_callb(sd_varlink *v, const char *method, sd_json_variant **ret_pa * 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); +int sd_varlink_call_and_upgradeb(sd_varlink *v, const char *method, sd_json_variant **ret_parameters, const char **ret_error_id, int *ret_input_fd, int *ret_output_fd, ...); +#define sd_varlink_call_and_upgradebo(v, method, ret_parameters, ret_error_id, ret_input_fd, ret_output_fd, ...) \ + sd_varlink_call_and_upgradeb((v), (method), (ret_parameters), (ret_error_id), (ret_input_fd), (ret_output_fd), SD_JSON_BUILD_OBJECT(__VA_ARGS__)) /* 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);