diff --git a/src/libsystemd/sd-json/json-stream.c b/src/libsystemd/sd-json/json-stream.c index ed4b6ca5b1..defda4a414 100644 --- a/src/libsystemd/sd-json/json-stream.c +++ b/src/libsystemd/sd-json/json-stream.c @@ -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); diff --git a/src/libsystemd/sd-varlink/sd-varlink.c b/src/libsystemd/sd-varlink/sd-varlink.c index a86e523a47..15c250fec3 100644 --- a/src/libsystemd/sd-varlink/sd-varlink.c +++ b/src/libsystemd/sd-varlink/sd-varlink.c @@ -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) diff --git a/src/libsystemd/sd-varlink/test-varlink.c b/src/libsystemd/sd-varlink/test-varlink.c index 73b63c45d7..ca43bcde90 100644 --- a/src/libsystemd/sd-varlink/test-varlink.c +++ b/src/libsystemd/sd-varlink/test-varlink.c @@ -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); diff --git a/src/shared/qmp-client.c b/src/shared/qmp-client.c index 8de903de1a..89b8551126 100644 --- a/src/shared/qmp-client.c +++ b/src/shared/qmp-client.c @@ -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() diff --git a/src/test/test-qmp-client-qemu.c b/src/test/test-qmp-client-qemu.c index 813b9c5687..82d7129af9 100644 --- a/src/test/test-qmp-client-qemu.c +++ b/src/test/test-qmp-client-qemu.c @@ -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)); diff --git a/src/vmspawn/vmspawn-qmp.c b/src/vmspawn/vmspawn-qmp.c index 092f8dae44..dccf7309cc 100644 --- a/src/vmspawn/vmspawn-qmp.c +++ b/src/vmspawn/vmspawn-qmp.c @@ -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"); diff --git a/src/vmspawn/vmspawn.c b/src/vmspawn/vmspawn.c index f3501c3b66..725e90dbec 100644 --- a/src/vmspawn/vmspawn.c +++ b/src/vmspawn/vmspawn.c @@ -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)