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

sd-varlink: several fixlets (#43369)

Browse Source
This commit is contained in:
Yu Watanabe
2026-08-15 01:38:18 +09:00
committed by GitHub
parent f807a32125 63da977f64
commit e3a224a46d
7 changed files with 114 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/libsystemd/sd-json/json-stream.c
+28 -17
View File
@@ -199,7 +199,6 @@ int json_stream_connect_address(JsonStream *s, const char *address) {
r = connect_unix_path(sock_fd, AT_FDCWD, address);
} else
r = RET_NERRNO(connect(sock_fd, &sockaddr.sa, r));
if (r < 0) {
if (!IN_SET(r, -EAGAIN, -EINPROGRESS))
return json_stream_log_errno(s, r, "Failed to connect to %s: %m", address);
@@ -212,26 +211,28 @@ int json_stream_connect_address(JsonStream *s, const char *address) {
s->flags |= JSON_STREAM_CONNECTING;
}
int fd = TAKE_FD(sock_fd);
return json_stream_attach_fds(s, fd, fd);
r = json_stream_attach_fds(s, sock_fd, sock_fd);
if (r < 0)
return r;
TAKE_FD(sock_fd);
return 0;
}
int json_stream_attach_fds(JsonStream *s, int input_fd, int output_fd) {
struct stat st;
assert(s);
assert(input_fd >= 0);
assert(input_fd != s->input_fd);
assert(input_fd != s->output_fd);
assert(output_fd >= 0);
assert(output_fd != s->input_fd);
assert(output_fd != s->output_fd);
/* NB: input_fd and output_fd are donated to the JsonStream instance! */
/* NB: input_fd and output_fd are donated to the JsonStream instance on success. */
if (s->input_fd != s->output_fd) {
safe_close(s->input_fd);
safe_close(s->output_fd);
} else
safe_close(s->input_fd);
s->input_fd = input_fd;
s->output_fd = output_fd;
s->flags &= ~(JSON_STREAM_PREFER_READ|JSON_STREAM_PREFER_WRITE);
JsonStreamFlags flags = s->flags & ~(JSON_STREAM_PREFER_READ|JSON_STREAM_PREFER_WRITE);
/* Detect non-socket fds up front so the read/write paths use read()/write() for
* non-socket fds and send()/recv() for sockets (mostly for MSG_NOSIGNAL). */
@@ -239,17 +240,24 @@ int json_stream_attach_fds(JsonStream *s, int input_fd, int output_fd) {
if (fstat(input_fd, &st) < 0)
return -errno;
if (!S_ISSOCK(st.st_mode))
s->flags |= JSON_STREAM_PREFER_READ;
flags |= JSON_STREAM_PREFER_READ;
}
if (output_fd >= 0 && output_fd != input_fd) {
if (fstat(output_fd, &st) < 0)
return -errno;
if (!S_ISSOCK(st.st_mode))
s->flags |= JSON_STREAM_PREFER_WRITE;
} else if (FLAGS_SET(s->flags, JSON_STREAM_PREFER_READ))
s->flags |= JSON_STREAM_PREFER_WRITE;
flags |= JSON_STREAM_PREFER_WRITE;
} else if (FLAGS_SET(flags, JSON_STREAM_PREFER_READ))
flags |= JSON_STREAM_PREFER_WRITE;
if (s->input_fd != s->output_fd)
safe_close(s->output_fd);
safe_close(s->input_fd);
s->input_fd = input_fd;
s->output_fd = output_fd;
s->flags = flags;
return 0;
}
@@ -260,6 +268,9 @@ int json_stream_connect_fd_pair(JsonStream *s, int input_fd, int output_fd) {
assert(input_fd >= 0);
assert(output_fd >= 0);
/* NB: input_fd and output_fd are donated to the JsonStream instance on success.
* The fds may be changed to non-blocking mode even on failure. */
r = fd_nonblock(input_fd, true);
if (r < 0)
return json_stream_log_errno(s, r, "Failed to make input fd %d nonblocking: %m", input_fd);
src/libsystemd/sd-varlink/sd-varlink.c
+29 -20
View File
@@ -302,20 +302,21 @@ _public_ int sd_varlink_connect_exec(sd_varlink **ret, const char *_command, cha
server_fd = safe_close(server_fd);
sd_varlink *v;
_cleanup_(sd_varlink_unrefp) sd_varlink *v = NULL;
r = varlink_new(&v);
if (r < 0)
return log_debug_errno(r, "Failed to create varlink object: %m");
int conn_fd = TAKE_FD(client_fd);
r = json_stream_attach_fds(&v->stream, conn_fd, conn_fd);
r = json_stream_attach_fds(&v->stream, client_fd, client_fd);
if (r < 0)
return r;
TAKE_FD(client_fd);
v->exec_pidref = TAKE_PIDREF(pidref);
varlink_set_state(v, VARLINK_IDLE_CLIENT);
*ret = v;
*ret = TAKE_PTR(v);
return 0;
}
@@ -390,20 +391,21 @@ static int varlink_connect_ssh_unix(sd_varlink **ret, const char *where) {
server_fd = safe_close(server_fd);
sd_varlink *v;
_cleanup_(sd_varlink_unrefp) sd_varlink *v = NULL;
r = varlink_new(&v);
if (r < 0)
return log_debug_errno(r, "Failed to create varlink object: %m");
int conn_fd = TAKE_FD(client_fd);
r = json_stream_attach_fds(&v->stream, conn_fd, conn_fd);
r = json_stream_attach_fds(&v->stream, client_fd, client_fd);
if (r < 0)
return r;
TAKE_FD(client_fd);
v->exec_pidref = TAKE_PIDREF(pidref);
varlink_set_state(v, VARLINK_IDLE_CLIENT);
*ret = v;
*ret = TAKE_PTR(v);
return 0;
}
@@ -485,19 +487,22 @@ static int varlink_connect_ssh_exec(sd_varlink **ret, const char *where) {
if (r < 0)
return log_debug_errno(r, "Failed to make output pipe non-blocking: %m");
sd_varlink *v;
_cleanup_(sd_varlink_unrefp) sd_varlink *v = NULL;
r = varlink_new(&v);
if (r < 0)
return log_debug_errno(r, "Failed to create varlink object: %m");
r = json_stream_attach_fds(&v->stream, TAKE_FD(output_pipe[0]), TAKE_FD(input_pipe[1]));
r = json_stream_attach_fds(&v->stream, output_pipe[0], input_pipe[1]);
if (r < 0)
return r;
TAKE_FD(output_pipe[0]);
TAKE_FD(input_pipe[1]);
v->exec_pidref = TAKE_PIDREF(pidref);
varlink_set_state(v, VARLINK_IDLE_CLIENT);
*ret = v;
*ret = TAKE_PTR(v);
return 0;
}
@@ -1768,7 +1773,7 @@ _public_ int sd_varlink_flush(sd_varlink *v) {
return json_stream_flush(&v->stream);
}
static void varlink_detach_server(sd_varlink *v) {
static void varlink_detach_server(sd_varlink *v, bool connected) {
sd_varlink_server *saved_server;
assert(v);
@@ -1803,7 +1808,7 @@ static void varlink_detach_server(sd_varlink *v) {
saved_server = TAKE_PTR(v->server);
if (saved_server->disconnect_callback)
if (saved_server->disconnect_callback && connected)
saved_server->disconnect_callback(saved_server, v, saved_server->userdata);
varlink_server_test_exit_on_idle(saved_server);
@@ -1817,12 +1822,14 @@ _public_ int sd_varlink_close(sd_varlink *v) {
if (v->state == VARLINK_DISCONNECTED)
return 0;
bool connected = v->state >= 0;
varlink_set_state(v, VARLINK_DISCONNECTED);
/* Let's take a reference first, since varlink_detach_server() might drop the final (dangling) ref
* which would destroy us before we can call varlink_clear() */
sd_varlink_ref(v);
varlink_detach_server(v);
varlink_detach_server(v, connected);
varlink_clear(v);
sd_varlink_unref(v);
@@ -3433,9 +3440,14 @@ _public_ int sd_varlink_server_add_connection_pair(
v->server = sd_varlink_server_ref(server);
sd_varlink_ref(v);
if (ucred_acquired)
json_stream_set_peer_ucred(&v->stream, &ucred);
r = json_stream_attach_fds(&v->stream, input_fd, output_fd);
if (r < 0)
if (r < 0) {
sd_varlink_close(v);
return r;
}
if (server->flags & SD_VARLINK_SERVER_INHERIT_USERDATA)
v->userdata = server->userdata;
@@ -3445,9 +3457,6 @@ _public_ int sd_varlink_server_add_connection_pair(
if (FLAGS_SET(server->flags, SD_VARLINK_SERVER_UPGRADABLE))
json_stream_set_flags(&v->stream, JSON_STREAM_BOUNDED_READS, true);
if (ucred_acquired)
json_stream_set_peer_ucred(&v->stream, &ucred);
_cleanup_free_ char *desc = NULL;
if (asprintf(&desc, "%s-%i-%i", varlink_server_description(server), input_fd, output_fd) >= 0)
json_stream_set_description(&v->stream, desc);
@@ -3455,8 +3464,6 @@ _public_ int sd_varlink_server_add_connection_pair(
(void) sd_varlink_set_allow_fd_passing_input(v, FLAGS_SET(server->flags, SD_VARLINK_SERVER_ALLOW_FD_PASSING_INPUT));
(void) sd_varlink_set_allow_fd_passing_output(v, FLAGS_SET(server->flags, SD_VARLINK_SERVER_ALLOW_FD_PASSING_OUTPUT));
varlink_set_state(v, VARLINK_IDLE_SERVER);
if (server->event) {
r = sd_varlink_attach_event(v, server->event, server->event_priority);
if (r < 0) {
@@ -3472,6 +3479,8 @@ _public_ int sd_varlink_server_add_connection_pair(
}
}
varlink_set_state(v, VARLINK_IDLE_SERVER);
(void) mark_varlink_socket(input_fd, /* path= */ NULL, "server");
if (ret)
src/libsystemd/sd-varlink/test-varlink.c
+43
View File
@@ -1504,4 +1504,47 @@ TEST(ctrunc) {
ASSERT_OK(r);
}
static void disconnect_callback(sd_varlink_server *server, sd_varlink *link, void *userdata) {
unsigned *counter = ASSERT_PTR(userdata);
(*counter)++;
}
TEST(add_connection_failure_cleanup) {
_cleanup_(sd_varlink_server_unrefp) sd_varlink_server *s = NULL;
_cleanup_close_pair_ int fds[2] = EBADF_PAIR;
const struct ucred ucred = {
.pid = getpid(),
.uid = getuid(),
.gid = getgid(),
};
ASSERT_OK(sd_varlink_server_new(&s, SD_VARLINK_SERVER_ACCOUNT_UID));
unsigned c = 0;
ASSERT_OK(sd_varlink_server_bind_disconnect(s, disconnect_callback));
ASSERT_NULL(sd_varlink_server_set_userdata(s, &c));
ASSERT_OK(sd_varlink_server_set_connections_per_uid_max(s, 1));
ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds));
int bad_fd = ASSERT_OK(memfd_new("closed-fd"));
ASSERT_OK_ERRNO(close(bad_fd));
ASSERT_ERROR(sd_varlink_server_add_connection_pair(s, fds[0], bad_fd, &ucred, /* ret= */ NULL), EBADF);
ASSERT_EQ(sd_varlink_server_current_connections(s), 0U);
ASSERT_EQ(c, 0U); /* The disconnect callback is not called. */
sd_varlink *v = NULL;
ASSERT_OK(sd_varlink_server_add_connection(s, fds[0], &v));
TAKE_FD(fds[0]);
ASSERT_NOT_NULL(v);
ASSERT_EQ(sd_varlink_server_current_connections(s), 1U);
ASSERT_EQ(c, 0U); /* The disconnect callback is not called yet. */
ASSERT_OK_POSITIVE(sd_varlink_close(v));
ASSERT_EQ(sd_varlink_server_current_connections(s), 0U);
ASSERT_EQ(c, 1U); /* The disconnect callback is called when the connection is closed. */
}
DEFINE_TEST_MAIN(LOG_DEBUG);
src/shared/qmp-client.c
+6 -2
View File
@@ -655,10 +655,12 @@ static int qmp_client_capabilities_reply(
int qmp_client_connect_fd(QmpClient **ret, int fd) {
_cleanup_(qmp_client_unrefp) QmpClient *c = NULL;
_cleanup_close_ int fd_close = ASSERT_FD(TAKE_FD(fd));
int r;
assert(ret);
assert(fd >= 0);
/* This function always takes ownership of the passed fd. */
c = new(QmpClient, 1);
if (!c)
@@ -681,10 +683,12 @@ int qmp_client_connect_fd(QmpClient **ret, int fd) {
if (r < 0)
return r;
r = json_stream_connect_fd_pair(&c->stream, fd, fd);
r = json_stream_connect_fd_pair(&c->stream, fd_close, fd_close);
if (r < 0)
return r;
TAKE_FD(fd_close);
/* Eagerly queue qmp_capabilities. QEMU accepts commands as soon as the socket opens
* — its greeting is informational and doesn't gate writes on our side. FIFO ordering
* of the output queue guarantees cap precedes any user command a later invoke()
src/test/test-qmp-client-qemu.c
+3 -6
View File
@@ -123,12 +123,11 @@ TEST(qmp_client_qemu_handshake_and_schema) {
ASSERT_OK(start_qemu(qemu, qmp_fds[1], &pidref));
qmp_fds[1] = safe_close(qmp_fds[1]);
r = qmp_client_connect_fd(&client, qmp_fds[0]);
r = qmp_client_connect_fd(&client, TAKE_FD(qmp_fds[0]));
if (r < 0) {
log_tests_skipped_errno(r, "QMP connect failed (QEMU may not support -machine none)");
return;
}
TAKE_FD(qmp_fds[0]);
ASSERT_OK(qmp_client_attach_event(client, event, SD_EVENT_PRIORITY_NORMAL));
@@ -187,12 +186,11 @@ TEST(qmp_client_qemu_query_status) {
ASSERT_OK(start_qemu(qemu, qmp_fds[1], &pidref));
qmp_fds[1] = safe_close(qmp_fds[1]);
r = qmp_client_connect_fd(&client, qmp_fds[0]);
r = qmp_client_connect_fd(&client, TAKE_FD(qmp_fds[0]));
if (r < 0) {
log_tests_skipped_errno(r, "QMP connect failed");
return;
}
TAKE_FD(qmp_fds[0]);
ASSERT_OK(qmp_client_attach_event(client, event, SD_EVENT_PRIORITY_NORMAL));
@@ -277,12 +275,11 @@ TEST(qmp_client_qemu_add_fd) {
ASSERT_OK(start_qemu(qemu, qmp_fds[1], &pidref));
qmp_fds[1] = safe_close(qmp_fds[1]);
r = qmp_client_connect_fd(&client, qmp_fds[0]);
r = qmp_client_connect_fd(&client, TAKE_FD(qmp_fds[0]));
if (r < 0) {
log_tests_skipped_errno(r, "QMP connect failed");
return;
}
TAKE_FD(qmp_fds[0]);
ASSERT_OK(qmp_client_attach_event(client, event, SD_EVENT_PRIORITY_NORMAL));
src/vmspawn/vmspawn-qmp.c
+4 -2
View File
@@ -1863,19 +1863,21 @@ static int probe_schema(QmpClient *c, VmspawnQmpBridge *bridge) {
int vmspawn_qmp_init(VmspawnQmpBridge **ret, int fd, sd_event *event) {
_cleanup_(vmspawn_qmp_bridge_freep) VmspawnQmpBridge *bridge = NULL;
_cleanup_close_ int fd_close = ASSERT_FD(TAKE_FD(fd));
int r;
assert(ret);
assert(fd >= 0);
assert(event);
/* This function always takes ownership of the passed fd. */
bridge = new0(VmspawnQmpBridge, 1);
if (!bridge)
return log_oom();
bridge->scsi_controller_port_idx = -1;
r = qmp_client_connect_fd(&bridge->qmp, fd);
r = qmp_client_connect_fd(&bridge->qmp, TAKE_FD(fd_close));
if (r < 0)
return log_error_errno(r, "Failed to create QMP client: %m");
src/vmspawn/vmspawn.c
+1 -3
View File
@@ -3865,12 +3865,10 @@ static int run_virtual_machine(int kvm_device_fd, int vhost_device_fd) {
/* Connect to VMM backend */
_cleanup_(vmspawn_qmp_bridge_freep) VmspawnQmpBridge *bridge = NULL;
r = vmspawn_qmp_init(&bridge, bridge_fds[0], event);
r = vmspawn_qmp_init(&bridge, TAKE_FD(bridge_fds[0]), event);
if (r < 0)
return r;
TAKE_FD(bridge_fds[0]);
/* Probe QEMU feature availability synchronously before device setup consumes the flags. */
r = vmspawn_qmp_probe_features(bridge);
if (r < 0)