From 35cadacd2bb9383686753731e31bd7e145fb2506 Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sat, 27 Apr 2024 06:24:32 +0200 Subject: Merging upstream version 10.0. Signed-off-by: Daniel Baumann --- lib/mgmt_msg.c | 107 +++++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 78 insertions(+), 29 deletions(-) (limited to 'lib/mgmt_msg.c') diff --git a/lib/mgmt_msg.c b/lib/mgmt_msg.c index 1cd3240..aff9af7 100644 --- a/lib/mgmt_msg.c +++ b/lib/mgmt_msg.c @@ -7,12 +7,15 @@ * Copyright (c) 2023, LabN Consulting, L.L.C. */ #include +#include + #include "debug.h" #include "network.h" #include "sockopt.h" #include "stream.h" #include "frrevent.h" #include "mgmt_msg.h" +#include "mgmt_msg_native.h" #define MGMT_MSG_DBG(dbgtag, fmt, ...) \ @@ -84,7 +87,7 @@ enum mgmt_msg_rsched mgmt_msg_read(struct mgmt_msg_state *ms, int fd, */ assert(stream_get_getp(ms->ins) == 0); left = stream_get_endp(ms->ins); - while (left > (long)sizeof(struct mgmt_msg_hdr)) { + while (left > (ssize_t)sizeof(struct mgmt_msg_hdr)) { mhdr = (struct mgmt_msg_hdr *)(STREAM_DATA(ms->ins) + total); if (!MGMT_MSG_IS_MARKER(mhdr->marker)) { MGMT_MSG_DBG(dbgtag, "recv corrupt buffer, disconnect"); @@ -99,8 +102,30 @@ enum mgmt_msg_rsched mgmt_msg_read(struct mgmt_msg_state *ms, int fd, mcount++; } - if (!mcount) + if (!mcount) { + /* Didn't manage to read a full message */ + if (mhdr && avail == 0) { + struct stream *news; + /* + * Message was longer than what was left and we have no + * available space to read more in. B/c mcount == 0 the + * message starts at the beginning of the stream so + * therefor the stream is too small to fit the message.. + * Resize the stream to fit. + */ + if (mhdr->len > MGMT_MSG_MAX_MSG_ALLOC_LEN) { + MGMT_MSG_ERR(ms, "corrupt msg len rcvd %u", + mhdr->len); + return MSR_DISCONNECT; + } + news = stream_new(mhdr->len); + stream_put(news, mhdr, left); + stream_set_endp(news, left); + stream_free(ms->ins); + ms->ins = news; + } return MSR_SCHED_STREAM; + } /* * We have read at least one message into the stream, queue it up. @@ -108,7 +133,11 @@ enum mgmt_msg_rsched mgmt_msg_read(struct mgmt_msg_state *ms, int fd, mhdr = (struct mgmt_msg_hdr *)(STREAM_DATA(ms->ins) + total); stream_set_endp(ms->ins, total); stream_fifo_push(&ms->inq, ms->ins); - ms->ins = stream_new(ms->max_msg_sz); + if (left < (ssize_t)sizeof(struct mgmt_msg_hdr)) + ms->ins = stream_new(ms->max_msg_sz); + else + /* handle case where message is greater than max */ + ms->ins = stream_new(MAX(ms->max_msg_sz, mhdr->len)); if (left) { stream_put(ms->ins, mhdr, left); stream_set_endp(ms->ins, left); @@ -180,7 +209,7 @@ bool mgmt_msg_procbufs(struct mgmt_msg_state *ms, } /** - * Write data from a onto the socket, using streams that have been queued for + * Write data onto the socket, using streams that have been queued for * sending by mgmt_msg_send_msg. This function should be reschedulable. * * Args: @@ -292,23 +321,26 @@ int mgmt_msg_send_msg(struct mgmt_msg_state *ms, uint8_t version, void *msg, size_t endp, n; size_t mlen = len + sizeof(*mhdr); - if (mlen > ms->max_msg_sz) { - MGMT_MSG_ERR(ms, "Message %zu > max size %zu, dropping", mlen, - ms->max_msg_sz); - return -1; - } + if (mlen > ms->max_msg_sz) + MGMT_MSG_DBG(dbgtag, "Sending large msg size %zu > max size %zu", + mlen, ms->max_msg_sz); if (!ms->outs) { - MGMT_MSG_DBG(dbgtag, "creating new stream for msg len %zu", - len); - ms->outs = stream_new(ms->max_msg_sz); + MGMT_MSG_DBG(dbgtag, "creating new stream for msg len %zu", mlen); + ms->outs = stream_new(MAX(ms->max_msg_sz, mlen)); + } else if (mlen > ms->max_msg_sz && ms->outs->endp == 0) { + /* msg is larger than stream max size get a fit-to-size stream */ + MGMT_MSG_DBG(dbgtag, + "replacing old stream with fit-to-size for msg len %zu", + mlen); + stream_free(ms->outs); + ms->outs = stream_new(mlen); } else if (STREAM_WRITEABLE(ms->outs) < mlen) { - MGMT_MSG_DBG( - dbgtag, - "enq existing stream len %zu and creating new stream for msg len %zu", - STREAM_WRITEABLE(ms->outs), mlen); + MGMT_MSG_DBG(dbgtag, + "enq existing stream len %zu and creating new stream for msg len %zu", + STREAM_WRITEABLE(ms->outs), mlen); stream_fifo_push(&ms->outq, ms->outs); - ms->outs = stream_new(ms->max_msg_sz); + ms->outs = stream_new(MAX(ms->max_msg_sz, mlen)); } else { MGMT_MSG_DBG( dbgtag, @@ -317,6 +349,16 @@ int mgmt_msg_send_msg(struct mgmt_msg_state *ms, uint8_t version, void *msg, } s = ms->outs; + if (dbgtag && version == MGMT_MSG_VERSION_NATIVE) { + struct mgmt_msg_header *native_msg = msg; + + MGMT_MSG_DBG( + dbgtag, + "Sending native msg sess/txn-id %"PRIu64" req-id %"PRIu64" code %u", + native_msg->refer_id, native_msg->req_id, native_msg->code); + + } + /* We have a stream with space, pack the message into it. */ mhdr = (struct mgmt_msg_hdr *)(STREAM_DATA(s) + s->endp); mhdr->marker = MGMT_MSG_MARKER(version); @@ -636,13 +678,13 @@ static void msg_client_sched_connect(struct msg_client *client, &client->conn_retry_tmr); } -static bool msg_client_connect_short_circuit(struct msg_client *client) +static int msg_client_connect_short_circuit(struct msg_client *client) { struct msg_conn *server_conn; struct msg_server *server; const char *dbgtag = client->conn.debug ? client->conn.mstate.idtag : NULL; - union sockunion su = {0}; + union sockunion su = {}; int sockets[2]; frr_each (msg_server_list, &msg_servers, server) @@ -650,10 +692,9 @@ static bool msg_client_connect_short_circuit(struct msg_client *client) break; if (!server) { MGMT_MSG_DBG(dbgtag, - "no short-circuit connection available for %s", + "no short-circuit server available yet for %s", client->sopath); - - return false; + return -1; } if (socketpair(AF_UNIX, SOCK_STREAM, 0, sockets)) { @@ -661,7 +702,7 @@ static bool msg_client_connect_short_circuit(struct msg_client *client) &client->conn.mstate, "socketpair failed trying to short-circuit connection on %s: %s", client->sopath, safe_strerror(errno)); - return false; + return -1; } /* client side */ @@ -673,6 +714,9 @@ static bool msg_client_connect_short_circuit(struct msg_client *client) /* server side */ memset(&su, 0, sizeof(union sockunion)); server_conn = server->create(sockets[1], &su); + server_conn->debug = DEBUG_MODE_CHECK(server->debug, DEBUG_MODE_ALL) + ? true + : false; client->conn.remote_conn = server_conn; server_conn->remote_conn = &client->conn; @@ -689,7 +733,7 @@ static bool msg_client_connect_short_circuit(struct msg_client *client) client->sopath, client->conn.mstate.idtag, client->conn.fd, server_conn->mstate.idtag, server_conn->fd); - return true; + return 0; } @@ -699,11 +743,12 @@ static void msg_client_connect(struct msg_client *client) struct msg_conn *conn = &client->conn; const char *dbgtag = conn->debug ? conn->mstate.idtag : NULL; - if (!client->short_circuit_ok || - !msg_client_connect_short_circuit(client)) + if (!client->short_circuit_ok) conn->fd = mgmt_msg_connect(client->sopath, MSG_CONN_SEND_BUF_SIZE, MSG_CONN_RECV_BUF_SIZE, dbgtag); + else if (msg_client_connect_short_circuit(client)) + conn->fd = -1; if (conn->fd == -1) /* retry the connection */ @@ -743,7 +788,6 @@ void msg_client_init(struct msg_client *client, struct event_loop *tm, mgmt_msg_init(&conn->mstate, max_read_buf, max_write_buf, max_msg_sz, idtag); - /* XXX maybe just have client kick this off */ /* Start trying to connect to server */ msg_client_sched_connect(client, 0); } @@ -766,8 +810,9 @@ void msg_client_cleanup(struct msg_client *client) static void msg_server_accept(struct event *event) { struct msg_server *server = EVENT_ARG(event); - int fd; + struct msg_conn *conn; union sockunion su; + int fd; if (server->fd < 0) return; @@ -790,7 +835,11 @@ static void msg_server_accept(struct event *event) DEBUGD(server->debug, "Accepted new %s connection", server->idtag); - server->create(fd, &su); + conn = server->create(fd, &su); + if (conn) + conn->debug = DEBUG_MODE_CHECK(server->debug, DEBUG_MODE_ALL) + ? true + : false; } int msg_server_init(struct msg_server *server, const char *sopath, -- cgit v1.2.3