json-stream: preserve message boundaries when receiving file descriptors

An AF_UNIX stream receive can combine a message without file descriptors
with a following SCM_RIGHTS-bearing message. This associates the received
descriptors with the earlier message even when the sender uses separate
sendmsg() calls.

Use the existing boundary-limited read path whenever FD input is enabled
on stream sockets, not only for protocol upgrades. Include the buffered
delimiter suffix in the peek search so a split CRLF delimiter cannot
over-read into the next message either.

Record the input socket type when attaching the descriptors so the new
FD-triggered limit does not truncate packet sockets. Explicit bounded
reads retain their existing semantics.

Add regressions for a oneway request followed by an FD-bearing request,
a continued reply followed by an FD-bearing terminal reply, a split CRLF
delimiter followed by an FD-bearing message, and batched messages on
SOCK_SEQPACKET and SOCK_DGRAM.

Fixes #43823
This commit is contained in:
Totoro
2026-09-24 09:05:42 +02:00
committed by Daan De Meyer
parent 1f286f971e
commit 3fe57d03a3
4 changed files with 322 additions and 16 deletions
+28 -16
View File
@@ -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");
+1
View File
@@ -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 */
+177
View File
@@ -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
+116
View File
@@ -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, &params));
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, &params));
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 = {};