diff --git a/src/libsystemd/sd-json/json-stream.c b/src/libsystemd/sd-json/json-stream.c index 676f50b9e2..883b59a301 100644 --- a/src/libsystemd/sd-json/json-stream.c +++ b/src/libsystemd/sd-json/json-stream.c @@ -132,6 +132,7 @@ static void json_stream_clear(JsonStream *s) { s->output_fd = safe_close(s->output_fd); } else s->output_fd = s->input_fd = safe_close(s->input_fd); + s->input_socket_type = 0; s->peer_pidfd = safe_close(s->peer_pidfd); s->ucred_acquired = false; @@ -221,6 +222,7 @@ int json_stream_connect_address(JsonStream *s, const char *address) { int json_stream_attach_fds(JsonStream *s, int input_fd, int output_fd) { struct stat st; + int input_socket_type = 0, r; assert(s); assert(input_fd >= 0); @@ -241,6 +243,11 @@ int json_stream_attach_fds(JsonStream *s, int input_fd, int output_fd) { return -errno; if (!S_ISSOCK(st.st_mode)) flags |= JSON_STREAM_PREFER_READ; + else { + r = getsockopt_int(input_fd, SOL_SOCKET, SO_TYPE, &input_socket_type); + if (r < 0) + return r; + } } if (output_fd >= 0 && output_fd != input_fd) { @@ -257,6 +264,7 @@ int json_stream_attach_fds(JsonStream *s, int input_fd, int output_fd) { s->input_fd = input_fd; s->output_fd = output_fd; + s->input_socket_type = input_socket_type; s->flags = flags; return 0; } @@ -1033,12 +1041,9 @@ static int json_stream_format_queue(JsonStream *s) { * Stop here and let json_stream_write() drain the buffer first; the next write() * call will pull this item into a clean buffer. * - * Note: this only produces a difference on SOCK_SEQPACKET / SOCK_DGRAM, where - * each sendmsg() is its own datagram with its own SCM_RIGHTS cmsg. On AF_UNIX - * SOCK_STREAM the kernel absorbs a preceding non-scm skb forward into the - * next scm-bearing skb's recv, so per-sendmsg separation is invisible to the - * receiver anyway. Kept as cheap defensive sender hygiene that's necessary - * the moment a SEQPACKET/DGRAM consumer wires JsonStream up. */ + * On AF_UNIX SOCK_STREAM, separate sendmsg() calls alone are not sufficient: + * recvmsg() may combine preceding data with the next SCM_RIGHTS-bearing send. + * The receive side must also stop at message boundaries to keep the association. */ if (q->n_fds > 0 && s->output_buffer_size > 0) return 0; @@ -1174,14 +1179,18 @@ int json_stream_write(JsonStream *s) { /* ===== Read side ===== */ -/* In bounded-reads mode, peek at the socket data to find the delimiter and return a read - * size that won't consume past it. This prevents over-reading data that belongs to whatever - * protocol the socket is being handed off to. Falls back to byte-by-byte for non-socket fds - * where MSG_PEEK is not available. */ +/* Peek at the socket data to find the delimiter and return a read size that won't consume past it. + * Besides preserving data for protocol handoff, this keeps SCM_RIGHTS from a later message from being + * received with an earlier message that carries no file descriptors. Falls back to byte-by-byte for + * non-socket fds where MSG_PEEK is not available. */ static ssize_t json_stream_peek_message_boundary(JsonStream *s, void *p, size_t rs) { assert(s); - if (!FLAGS_SET(s->flags, JSON_STREAM_BOUNDED_READS)) + if ((s->flags & (JSON_STREAM_BOUNDED_READS|JSON_STREAM_ALLOW_FD_PASSING_INPUT)) == 0) + return rs; + + /* Packet sockets already preserve the FD boundary. Shortening their reads would truncate data. */ + if (s->input_socket_type != SOCK_STREAM && !FLAGS_SET(s->flags, JSON_STREAM_BOUNDED_READS)) return rs; if (FLAGS_SET(s->flags, JSON_STREAM_PREFER_READ)) @@ -1189,6 +1198,9 @@ static ssize_t json_stream_peek_message_boundary(JsonStream *s, void *p, size_t ssize_t peeked = recv(s->input_fd, p, rs, MSG_PEEK|MSG_DONTWAIT); if (peeked < 0) { + /* Let the actual read classify disconnections and update the stream state. */ + if (ERRNO_IS_DISCONNECT(errno)) + return rs; if (!ERRNO_IS_TRANSIENT(errno)) return -errno; @@ -1200,9 +1212,11 @@ static ssize_t json_stream_peek_message_boundary(JsonStream *s, void *p, size_t return rs; size_t dsz = json_stream_delimiter_size(s); - void *delim = memmem_safe(p, peeked, s->delimiter ?: "\0", dsz); + /* Include the buffered suffix in case the delimiter straddles the old and new data. */ + size_t rescan = MIN(s->input_buffer_size, dsz - 1); + void *delim = memmem_safe((char*) p - rescan, rescan + peeked, s->delimiter ?: "\0", dsz); if (delim) - return (ssize_t) ((char*) delim - (char*) p) + dsz; + return (char*) delim + dsz - (char*) p; return peeked; } @@ -1259,9 +1273,7 @@ int json_stream_read(JsonStream *s) { rs = MALLOC_SIZEOF_SAFE(s->input_buffer) - (s->input_buffer_index + s->input_buffer_size); - /* If a protocol upgrade may follow, ensure we don't consume any post-upgrade bytes by - * limiting the read to the next delimiter. Uses MSG_PEEK on sockets, single-byte reads - * otherwise. */ + /* Preserve message boundaries when receiving file descriptors or preparing a protocol upgrade. */ rs = json_stream_peek_message_boundary(s, p, rs); if (rs < 0) return json_stream_log_errno(s, (int) rs, "Failed to peek message boundary: %m"); diff --git a/src/libsystemd/sd-json/json-stream.h b/src/libsystemd/sd-json/json-stream.h index 87ad23397d..8af85610ff 100644 --- a/src/libsystemd/sd-json/json-stream.h +++ b/src/libsystemd/sd-json/json-stream.h @@ -75,6 +75,7 @@ typedef struct JsonStream { int input_fd; int output_fd; + int input_socket_type; /* Cached SO_TYPE, or 0 for non-socket input. */ usec_t timeout; /* relative; USEC_INFINITY = no timeout */ usec_t last_activity; /* CLOCK_MONOTONIC */ diff --git a/src/libsystemd/sd-varlink/test-varlink.c b/src/libsystemd/sd-varlink/test-varlink.c index ca43bcde90..169a5f372b 100644 --- a/src/libsystemd/sd-varlink/test-varlink.c +++ b/src/libsystemd/sd-varlink/test-varlink.c @@ -20,7 +20,9 @@ #include "path-util.h" #include "rm-rf.h" #include "socket-util.h" +#include "string-util.h" #include "tests.h" +#include "time-util.h" #include "tmpfile-util.h" #include "varlink-util.h" @@ -695,6 +697,181 @@ TEST(sentinel_oneway) { ASSERT_OK(sd_event_loop(e)); } +typedef struct FdAssociationState { + int first_fd; + int second_fd; + unsigned n_messages; +} FdAssociationState; + +static int method_fd_association_oneway( + sd_varlink *link, + sd_json_variant *parameters, + sd_varlink_method_flags_t flags, + void *userdata) { + + FdAssociationState *state = ASSERT_PTR(userdata); + + ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_ONEWAY)); + state->first_fd = sd_varlink_peek_fd(link, 0); + state->n_messages++; + return 0; +} + +static int method_fd_association_receive( + sd_varlink *link, + sd_json_variant *parameters, + sd_varlink_method_flags_t flags, + void *userdata) { + + FdAssociationState *state = ASSERT_PTR(userdata); + + state->second_fd = sd_varlink_peek_fd(link, 0); + if (state->second_fd >= 0) + test_fd(state->second_fd, "second-message", STRLEN("second-message")); + state->n_messages++; + + /* Reply even when the FD is missing, so the assertions below report the association failure. */ + return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_BOOLEAN("received", state->second_fd >= 0)); +} + +static int reply_fd_association_done( + sd_varlink *link, + sd_json_variant *parameters, + const char *error_id, + sd_varlink_reply_flags_t flags, + void *userdata) { + + ASSERT_NULL(error_id); + return sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS); +} + +static void run_fd_association_after_oneway(bool bounded) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *server = NULL; + _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *client = NULL; + _cleanup_(sd_event_unrefp) sd_event *event = NULL; + _cleanup_close_pair_ int sockets[2] = EBADF_PAIR; + _cleanup_close_ int fd = -EBADF; + FdAssociationState state = { .first_fd = -EBADF, .second_fd = -EBADF }; + + ASSERT_OK(sd_event_new(&event)); + ASSERT_OK(sd_varlink_server_new( + &server, + SD_VARLINK_SERVER_ALLOW_FD_PASSING_INPUT|SD_VARLINK_SERVER_INHERIT_USERDATA| + (bounded ? SD_VARLINK_SERVER_UPGRADABLE : 0))); + sd_varlink_server_set_userdata(server, &state); + ASSERT_OK(sd_varlink_server_attach_event(server, event, 1)); + ASSERT_OK(sd_varlink_server_bind_method_many( + server, + "io.test.OnewayPulse", method_fd_association_oneway, + "io.test.DeliverFD", method_fd_association_receive)); + + ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, sockets)); + ASSERT_OK(sd_varlink_server_add_connection(server, sockets[0], /* ret= */ NULL)); + TAKE_FD(sockets[0]); + ASSERT_OK(sd_varlink_connect_fd(&client, sockets[1])); + TAKE_FD(sockets[1]); + ASSERT_OK(sd_varlink_set_allow_fd_passing_output(client, true)); + ASSERT_OK(sd_varlink_set_relative_timeout(client, 5 * USEC_PER_SEC)); + ASSERT_OK(sd_varlink_attach_event(client, event, 0)); + ASSERT_OK(sd_varlink_bind_reply(client, reply_fd_association_done)); + + /* Queue both messages before the server runs: only the second message owns the descriptor. */ + ASSERT_OK(sd_varlink_send(client, "io.test.OnewayPulse", /* parameters= */ NULL)); + ASSERT_OK(fd = memfd_new_and_seal_string("association", "second-message")); + ASSERT_OK_EQ(sd_varlink_push_dup_fd(client, fd), 0); + ASSERT_OK(sd_varlink_invoke(client, "io.test.DeliverFD", /* parameters= */ NULL)); + ASSERT_OK(sd_varlink_flush(client)); + ASSERT_OK(sd_event_loop(event)); + + log_info("Request FD association (bounded=%s): first=%d second=%d messages=%u", + yes_no(bounded), state.first_fd, state.second_fd, state.n_messages); + ASSERT_EQ(state.n_messages, 2U); + ASSERT_ERROR(state.first_fd, ENXIO); + ASSERT_OK(state.second_fd); +} + +TEST(fd_association_after_oneway_bounded) { + /* An upgradable server already uses boundary-limited reads; keep this as a diagnostic control. */ + run_fd_association_after_oneway(true); +} + +TEST(fd_association_after_oneway) { + run_fd_association_after_oneway(false); +} + +static int method_fd_association_reply( + sd_varlink *link, + sd_json_variant *parameters, + sd_varlink_method_flags_t flags, + void *userdata) { + + _cleanup_close_ int fd = -EBADF; + + ASSERT_TRUE(FLAGS_SET(flags, SD_VARLINK_METHOD_MORE)); + ASSERT_OK(sd_varlink_notifybo(link, SD_JSON_BUILD_PAIR_UNSIGNED("sequence", 0))); + ASSERT_OK(fd = memfd_new_and_seal_string("association", "second-message")); + ASSERT_OK_EQ(sd_varlink_push_dup_fd(link, fd), 0); + return sd_varlink_replybo(link, SD_JSON_BUILD_PAIR_UNSIGNED("sequence", 1)); +} + +static int reply_fd_association_more( + sd_varlink *link, + sd_json_variant *parameters, + const char *error_id, + sd_varlink_reply_flags_t flags, + void *userdata) { + + FdAssociationState *state = ASSERT_PTR(userdata); + + ASSERT_NULL(error_id); + if (FLAGS_SET(flags, SD_VARLINK_REPLY_CONTINUES)) { + ASSERT_EQ(state->n_messages, 0U); + state->first_fd = sd_varlink_peek_fd(link, 0); + state->n_messages++; + return 0; + } + + ASSERT_EQ(state->n_messages, 1U); + state->second_fd = sd_varlink_peek_fd(link, 0); + if (state->second_fd >= 0) + test_fd(state->second_fd, "second-message", STRLEN("second-message")); + state->n_messages++; + return sd_event_exit(sd_varlink_get_event(link), EXIT_SUCCESS); +} + +TEST(fd_association_after_continued_reply) { + _cleanup_(sd_varlink_server_unrefp) sd_varlink_server *server = NULL; + _cleanup_(sd_varlink_flush_close_unrefp) sd_varlink *client = NULL; + _cleanup_(sd_event_unrefp) sd_event *event = NULL; + _cleanup_close_pair_ int sockets[2] = EBADF_PAIR; + FdAssociationState state = { .first_fd = -EBADF, .second_fd = -EBADF }; + + ASSERT_OK(sd_event_new(&event)); + ASSERT_OK(sd_varlink_server_new(&server, SD_VARLINK_SERVER_ALLOW_FD_PASSING_OUTPUT)); + ASSERT_OK(sd_varlink_server_attach_event(server, event, 0)); + ASSERT_OK(sd_varlink_server_bind_method( + server, "io.test.DeliverReplies", method_fd_association_reply)); + ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_NONBLOCK|SOCK_CLOEXEC, 0, sockets)); + ASSERT_OK(sd_varlink_server_add_connection(server, sockets[0], /* ret= */ NULL)); + TAKE_FD(sockets[0]); + ASSERT_OK(sd_varlink_connect_fd(&client, sockets[1])); + TAKE_FD(sockets[1]); + sd_varlink_set_userdata(client, &state); + ASSERT_OK(sd_varlink_set_allow_fd_passing_input(client, true)); + ASSERT_OK(sd_varlink_set_relative_timeout(client, 5 * USEC_PER_SEC)); + ASSERT_OK(sd_varlink_attach_event(client, event, 1)); + ASSERT_OK(sd_varlink_bind_reply(client, reply_fd_association_more)); + ASSERT_OK(sd_varlink_observe(client, "io.test.DeliverReplies", /* parameters= */ NULL)); + ASSERT_OK(sd_varlink_flush(client)); + ASSERT_OK(sd_event_loop(event)); + + log_info("Reply FD association: first=%d second=%d messages=%u", + state.first_fd, state.second_fd, state.n_messages); + ASSERT_EQ(state.n_messages, 2U); + ASSERT_ERROR(state.first_fd, ENXIO); + ASSERT_OK(state.second_fd); +} + static int method_fiber_sentinel_error(sd_varlink *link, sd_json_variant *parameters, sd_varlink_method_flags_t flags, void *userdata) { /* Set an error sentinel from a fiber callback and return without sending a reply. The sentinel * error should still be propagated by the fiber's post-callback logic, even though the varlink diff --git a/src/test/test-qmp-client.c b/src/test/test-qmp-client.c index f13a8913ed..9f6429ea2c 100644 --- a/src/test/test-qmp-client.c +++ b/src/test/test-qmp-client.c @@ -10,8 +10,10 @@ #include "errno-util.h" #include "fd-util.h" +#include "iovec-util.h" #include "json-stream.h" #include "qmp-client.h" +#include "socket-util.h" #include "string-util.h" #include "tests.h" @@ -251,6 +253,120 @@ TEST(json_stream_split_crlf_delimiter) { ASSERT_FALSE(json_stream_has_buffered_input(&s)); } +TEST(json_stream_fd_passing_disconnect) { + _cleanup_(json_stream_done) JsonStream s = {}; + _cleanup_close_pair_ int fds[2] = EBADF_PAIR; + + ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds)); + mock_qmp_init(&s, TAKE_FD(fds[1])); + ASSERT_OK(json_stream_set_allow_fd_passing_input(&s, true, true)); + + /* Closing a peer with unread data causes ECONNRESET. The peek must leave disconnect + * classification to the actual recvmsg() call. */ + ASSERT_OK_EQ_ERRNO(write(s.output_fd, "x", 1), 1); + fds[0] = safe_close(fds[0]); + + ASSERT_OK_POSITIVE(json_stream_read(&s)); + ASSERT_TRUE(FLAGS_SET(s.flags, JSON_STREAM_READ_DISCONNECTED)); +} + +TEST(json_stream_fd_association_split_crlf_delimiter) { + const JsonStreamParams params = { + .delimiter = "\r\n", + .phase = mock_qmp_phase, + .dispatch = mock_qmp_dispatch, + }; + _cleanup_(json_stream_done) JsonStream s = {}; + _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL; + _cleanup_close_pair_ int fds[2] = EBADF_PAIR; + _cleanup_close_ int fd_to_pass = -EBADF; + const char first[] = "{\"first\":true}\r"; + const struct iovec second = CONST_IOVEC_MAKE_STRING("{\"second\":true}\r\n"); + eventfd_t value; + + ASSERT_OK_ERRNO(socketpair(AF_UNIX, SOCK_STREAM|SOCK_CLOEXEC, 0, fds)); + ASSERT_OK(json_stream_init(&s, ¶ms)); + int fd = TAKE_FD(fds[0]); + ASSERT_OK(json_stream_connect_fd_pair(&s, fd, fd)); + ASSERT_OK(json_stream_set_allow_fd_passing_input(&s, true, true)); + + /* Buffer the first message without the final byte of its delimiter. */ + ASSERT_OK_EQ_ERRNO(write(fds[1], first, strlen(first)), (ssize_t) strlen(first)); + ASSERT_OK_POSITIVE(json_stream_read(&s)); + ASSERT_OK_ZERO(json_stream_parse(&s, &v)); + ASSERT_NULL(v); + + /* The remaining newline must not bring the next message's descriptor into the first message. */ + ASSERT_OK_EQ_ERRNO(write(fds[1], "\n", 1), 1); + fd_to_pass = ASSERT_OK_ERRNO(eventfd(7, EFD_CLOEXEC|EFD_NONBLOCK)); + ASSERT_OK_EQ(send_one_fd_iov(fds[1], fd_to_pass, &second, 1, MSG_NOSIGNAL), + (ssize_t) second.iov_len); + ASSERT_OK_POSITIVE(json_stream_read(&s)); + ASSERT_OK_POSITIVE(json_stream_parse(&s, &v)); + ASSERT_TRUE(sd_json_variant_boolean(ASSERT_NOT_NULL(sd_json_variant_by_key(v, "first")))); + ASSERT_EQ(json_stream_get_n_input_fds(&s), (size_t) 0); + + v = sd_json_variant_unref(v); + ASSERT_OK_POSITIVE(json_stream_read(&s)); + ASSERT_OK_POSITIVE(json_stream_parse(&s, &v)); + ASSERT_TRUE(sd_json_variant_boolean(ASSERT_NOT_NULL(sd_json_variant_by_key(v, "second")))); + ASSERT_EQ(json_stream_get_n_input_fds(&s), (size_t) 1); + int received_fd = ASSERT_OK(json_stream_peek_input_fd(&s, 0)); + ASSERT_OK_ERRNO(eventfd_read(received_fd, &value)); + ASSERT_EQ(value, (eventfd_t) 7); + ASSERT_FALSE(json_stream_has_buffered_input(&s)); +} + +static void test_json_stream_packet_batch(int type) { + const JsonStreamParams params = { + .delimiter = "\r\n", + .phase = mock_qmp_phase, + .dispatch = mock_qmp_dispatch, + }; + _cleanup_(json_stream_done) JsonStream s = {}; + _cleanup_(sd_json_variant_unrefp) sd_json_variant *v = NULL; + _cleanup_close_pair_ int fds[2] = EBADF_PAIR; + _cleanup_close_ int fd_to_pass = -EBADF; + const struct iovec batch = CONST_IOVEC_MAKE_STRING("{\"first\":true}\r\n{\"second\":true}\r\n"); + eventfd_t value; + + ASSERT_OK_ERRNO(socketpair(AF_UNIX, type|SOCK_CLOEXEC, 0, fds)); + ASSERT_OK(json_stream_init(&s, ¶ms)); + int fd = TAKE_FD(fds[0]); + ASSERT_OK(json_stream_connect_fd_pair(&s, fd, fd)); + ASSERT_OK(json_stream_set_allow_fd_passing_input(&s, true, true)); + + /* Both messages share one packet, with the descriptor attached to the first message. + * Limiting recvmsg() to the first delimiter would truncate the rest of the packet. */ + fd_to_pass = ASSERT_OK_ERRNO(eventfd(7, EFD_CLOEXEC|EFD_NONBLOCK)); + ASSERT_OK_EQ(send_one_fd_iov(fds[1], fd_to_pass, &batch, 1, MSG_NOSIGNAL), + (ssize_t) batch.iov_len); + ASSERT_OK_POSITIVE(json_stream_read(&s)); + ASSERT_OK_POSITIVE(json_stream_parse(&s, &v)); + sd_json_variant *first = ASSERT_NOT_NULL(sd_json_variant_by_key(v, "first")); + ASSERT_TRUE(sd_json_variant_boolean(first)); + ASSERT_EQ(json_stream_get_n_input_fds(&s), (size_t) 1); + int received_fd = ASSERT_OK(json_stream_peek_input_fd(&s, 0)); + ASSERT_OK_ERRNO(eventfd_read(received_fd, &value)); + ASSERT_EQ(value, (eventfd_t) 7); + json_stream_close_input_fds(&s); + + v = sd_json_variant_unref(v); + ASSERT_OK_POSITIVE(json_stream_parse(&s, &v)); + sd_json_variant *second = ASSERT_NOT_NULL(sd_json_variant_by_key(v, "second")); + ASSERT_TRUE(sd_json_variant_boolean(second)); + ASSERT_EQ(json_stream_get_n_input_fds(&s), (size_t) 0); + ASSERT_FALSE(json_stream_has_buffered_input(&s)); +} + +TEST(json_stream_seqpacket_batch) { + test_json_stream_packet_batch(SOCK_SEQPACKET); +} + +TEST(json_stream_dgram_batch) { + test_json_stream_packet_batch(SOCK_DGRAM); +} + static int mock_qmp_eof_fiber(void *userdata) { _cleanup_(json_stream_done) JsonStream s = {};