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/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 90be017705..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);
@@ -1544,6 +1586,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 +1610,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
@@ -2385,6 +2433,56 @@ _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");
+ }
+
+ /* 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);
+ 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);
+ 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 +2534,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 +2553,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);
@@ -2806,6 +2867,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);
@@ -3600,7 +3752,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)
@@ -4557,6 +4710,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..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;
@@ -137,8 +138,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 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);
@@ -168,6 +170,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 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
+ * 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 +336,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..1bbc87c32c 100644
--- a/src/test/test-varlink.c
+++ b/src/test/test-varlink.c
@@ -11,9 +11,11 @@
#include "sd-varlink.h"
#include "fd-util.h"
+#include "io-util.h"
#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"
@@ -725,7 +727,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 +754,186 @@ 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;
+
+ /* 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]);
+
+ ASSERT_OK(loop_write(output_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);
+ ASSERT_NE(input_fd, output_fd); /* library dups for bidirectional sockets */
+
+ /* Send a test string, shut down write side so 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");
+
+ /* 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, 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));
+ 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));
+}
+
+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 00bd71f34d..18a639962c 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. */
@@ -1169,6 +1160,155 @@ 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 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");
+
+ /* 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|SD_VARLINK_SERVER_UPGRADABLE);
+ 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"