Validate cluster module message lengths before allocation

src/cluster.h

@@ -54,6 +54,8 @@ void clusterBeforeSleep(void);
 int verifyClusterConfigWithData(void);
 void clusterHandleServerShutdown(bool auto_failover);
 
+int clusterModuleMessageLengthValid(uint32_t len);
+
 int clusterSendModuleMessageToTarget(const char *target,
                                      uint64_t module_id,
                                      uint8_t type,

src/cluster_legacy.c

@@ -5671,6 +5671,16 @@ static inline bool nodeSupportsLightMsgHdrForModule(clusterNode *n) {
            (n->flags & CLUSTER_NODE_LIGHT_HDR_MODULE_SUPPORTED);
 }
 
+/* Return whether the payload fits in the cluster bus's 32-bit length field,
+ * including the normal message header and the in-memory send-block prefix.
+ * The normal header is larger than the light header, so this limit is safe
+ * regardless of the destination node's negotiated message format. */
+int clusterModuleMessageLengthValid(uint32_t len) {
+    size_t overhead = offsetof(clusterMsgSendBlock, data) + CLUSTERMSG_MIN_LEN +
+                      offsetof(clusterMsgModule, bulk_data);
+    return (size_t)len <= (size_t)UINT32_MAX - overhead;
+}
+
 /* Create a MODULE message block.
  *
  * If is_light is 1, then build a message block with `clusterMsgLight` struct else `clusterMsg`. */
@@ -5754,6 +5764,7 @@ int clusterSendModuleMessageToTarget(const char *target,
         if (node == NULL || node->link == NULL) return C_ERR;
     }
 
+    if (!clusterModuleMessageLengthValid(len)) return C_ERR;
     clusterSendModule(target ? node->link : NULL, module_id, type, payload, len);
     return C_OK;
 }

src/module.c

@@ -9982,7 +9982,8 @@ void VM_RegisterClusterMessageReceiver(ValkeyModuleCtx *ctx,
  *
  * The function returns VALKEYMODULE_OK if the message was successfully sent,
  * otherwise if the node is not connected or such node ID does not map to any
- * known cluster node, VALKEYMODULE_ERR is returned. */
+ * known cluster node, or if the payload exceeds the cluster bus message size
+ * limit, VALKEYMODULE_ERR is returned. */
 int VM_SendClusterMessage(ValkeyModuleCtx *ctx, const char *target_id, uint8_t type, const char *msg, uint32_t len) {
     if (!server.cluster_enabled) return VALKEYMODULE_ERR;
     uint64_t module_id = moduleTypeEncodeId(ctx->module->name, 0);

src/unit/test_cluster_module.cpp

@@ -0,0 +1,67 @@
+/*
+ * Copyright (c) Valkey Contributors
+ * All rights reserved.
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "generated_wrappers.hpp"
+
+#include <cstddef>
+#include <cstdint>
+
+extern "C" {
+#include "cluster.h"
+#include "cluster_legacy.h"
+#include "server.h"
+}
+
+/* Mirrors clusterMsgSendBlock, which is private to cluster_legacy.c. */
+typedef struct TestClusterMsgSendBlock {
+    size_t totlen;
+    int refcount;
+    union {
+        clusterMsg msg;
+        clusterMsgLight msg_light;
+    } data[1];
+} TestClusterMsgSendBlock;
+
+/* Returns the largest payload length that clusterModuleMessageLengthValid()
+ * accepts for a normal cluster header. The limit keeps the full send-block
+ * allocation within UINT32_MAX. */
+static uint32_t maxModulePayloadLen(void) {
+    size_t overhead = offsetof(TestClusterMsgSendBlock, data) + CLUSTERMSG_MIN_LEN +
+                      offsetof(clusterMsgModule, bulk_data);
+    return (uint32_t)((size_t)UINT32_MAX - overhead);
+}
+
+TEST(ClusterModuleTest, ValidatesFullAllocationLength) {
+    uint32_t max_payload_len = maxModulePayloadLen();
+
+    EXPECT_EQ(clusterModuleMessageLengthValid(0), 1);
+    EXPECT_EQ(clusterModuleMessageLengthValid(3), 1);
+    EXPECT_EQ(clusterModuleMessageLengthValid(max_payload_len), 1);
+    EXPECT_EQ(clusterModuleMessageLengthValid(max_payload_len + 1), 0);
+    EXPECT_EQ(clusterModuleMessageLengthValid(UINT32_MAX), 0);
+}
+
+TEST(ClusterModuleTest, NormalHeaderLimitAlsoProtectsLightHeaders) {
+    size_t normal_overhead = offsetof(TestClusterMsgSendBlock, data) + CLUSTERMSG_MIN_LEN +
+                             offsetof(clusterMsgModule, bulk_data);
+    size_t light_overhead = offsetof(TestClusterMsgSendBlock, data) + CLUSTERMSG_LIGHT_MIN_LEN +
+                            offsetof(clusterMsgModule, bulk_data);
+    uint32_t max_payload_len = maxModulePayloadLen();
+
+    EXPECT_GT(normal_overhead, light_overhead);
+    EXPECT_EQ((size_t)max_payload_len + normal_overhead, (size_t)UINT32_MAX);
+    EXPECT_LE((size_t)max_payload_len + light_overhead, (size_t)UINT32_MAX);
+}
+
+/* Reject oversized messages at the actual cluster-module send boundary.
+ * The tiny payload must never be accessed for an invalid length. */
+TEST(ClusterModuleTest, RejectsOversizedMessageBeforeSending) {
+    const char payload[] = "x";
+    uint32_t first_invalid = maxModulePayloadLen() + 1;
+
+    EXPECT_EQ(clusterSendModuleMessageToTarget(NULL, 0, 1, payload, first_invalid), C_ERR);
+    EXPECT_EQ(clusterSendModuleMessageToTarget(NULL, 0, 1, payload, UINT32_MAX), C_ERR);
+}

tests/modules/cluster.c

@@ -1,4 +1,5 @@
 #include "valkeymodule.h"
+#include <stdint.h>
 #include <string.h>
 
 #define UNUSED(x) (void)(x)
@@ -78,6 +79,17 @@ int PingallCommand(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int argc) {
     return ValkeyModule_ReplyWithSimpleString(ctx, "OK");
 }
 
+/* Test that an impossible size is rejected before reading the payload. */
+int test_send_oversized_msg(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int argc) {
+    UNUSED(argv);
+    if (argc != 1) return ValkeyModule_WrongArity(ctx);
+
+    if (ValkeyModule_SendClusterMessage(ctx, NULL, MSGTYPE_DING, "x", UINT32_MAX) != VALKEYMODULE_ERR) {
+        return ValkeyModule_ReplyWithError(ctx, "ERR oversized cluster module message was accepted");
+    }
+    return ValkeyModule_ReplyWithSimpleString(ctx, "REJECTED");
+}
+
 void DingReceiver(ValkeyModuleCtx *ctx, const char *sender_id, uint8_t type, const unsigned char *payload, uint32_t len) {
     ValkeyModule_Log(ctx, "notice", "DING (type %d) RECEIVED from %.*s: '%.*s'", type, VALKEYMODULE_NODE_ID_LEN, sender_id, (int)len, payload);
     /* Ensure sender_id is null-terminated for cross-version compatibility */
@@ -177,6 +189,10 @@ int ValkeyModule_OnLoad(ValkeyModuleCtx *ctx, ValkeyModuleString **argv, int arg
         VALKEYMODULE_ERR)
         return VALKEYMODULE_ERR;
 
+    if (ValkeyModule_CreateCommand(ctx, "test.send_oversized_msg", test_send_oversized_msg, "readonly", 0, 0, 0) ==
+        VALKEYMODULE_ERR)
+        return VALKEYMODULE_ERR;
+
     if (ValkeyModule_CreateCommand(ctx, "test.cluster_slots", test_cluster_slots, "", 0, 0, 0) == VALKEYMODULE_ERR)
         return VALKEYMODULE_ERR;
 

tests/unit/moduleapi/cluster.tcl

@@ -289,6 +289,15 @@ start_cluster 3 0 [list config_lines $modules] {
     set node2 [srv -1 client]
     set node3 [srv -2 client]
 
+    test "VM_SendClusterMessage rejects oversized payloads before reading them" {
+        R 0 CONFIG RESETSTAT
+        assert_equal {REJECTED} [$node1 test.send_oversized_msg]
+        set module_sent 0
+        regexp {cluster_stats_messages_module_sent:([0-9]+)} [$node1 CLUSTER INFO] -> module_sent
+        assert_equal {0} $module_sent
+        assert_equal {PONG} [$node1 PING]
+    }
+
     test "VM_CALL with cluster slots" {
         assert_equal [lsort [$node1 cluster slots]] [lsort [$node1 test.cluster_slots]]
         assert_equal [lsort [$node2 cluster slots]] [lsort [$node2 test.cluster_slots]]