Repository: valkey-io/valkey
File: src/cluster_legacy.c (C)

Code before:
 * received from the wrong sender ID). */
int clusterProcessPacket(clusterLink *link) {
    /* Validate that the packet is well-formed */
    if (!clusterIsValidPacket(link)) {
        clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
        uint16_t type = ntohs(hdr->type);
        if (server.debug_cluster_close_link_on_packet_drop &&
            (type == server.cluster_drop_packet_filter || server.cluster_drop_packet_filter == -2)) {
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %s", clusterGetMessageTypeString(type));
            return 0;
        }
        return 1;
    }

    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
    mstime_t now = mstime();
    int is_light = IS_LIGHT_MESSAGE(ntohs(hdr->type));
    uint16_t type = ntohs(hdr->type) & ~CLUSTERMSG_MODIFIER_MASK;

    if (is_light) {
        if (!link->node || nodeInHandshake(link->node)) {
            serverLog(
                LL_NOTICE,
                "Closing link for node %.40s that sent a lightweight message of type %s as its first message on the link",
                link->node ? link->node->name : "<unknown>",
                clusterGetMessageTypeString(type));
            freeClusterLink(link);
            return 0;
        }
        clusterNode *sender = link->node;
        sender->data_received = now;
        clusterProcessLightPacket(sender, link, type);
        return 1;
    }

    uint16_t flags = ntohs(hdr->flags);
    uint64_t sender_claimed_current_epoch = 0, sender_claimed_config_epoch = 0;
    clusterNode *sender = getNodeFromLinkAndMsg(link, hdr);
    int sender_claims_to_be_primary = !memcmp(hdr->replicaof, CLUSTER_NODE_NULL_NAME, CLUSTER_NAMELEN);
    int sender_last_reported_as_replica = sender && nodeIsReplica(sender);
    int sender_last_reported_as_primary = sender && nodeIsPrimary(sender);

    /* We store this information at the link layer so that we can send extensions
     * during the handshake even if we don't know the sender. */
    if (hdr->mflags[0] & CLUSTERMSG_FLAG0_EXT_DATA) {
        link->flags |= CLUSTER_LINK_EXTENSIONS_SUPPORTED;
    }

    /* Store some flags about the sender. */
    if (sender) {

Review comment (suggestion):
This change is fine, but it doesn't really address the root cause of the issue.  The root cause is that `link->rcvbuf` is being blindly cast to a `clusterMsg*` without actually checking the type flag.  Then, the variable is left unchecked for the code to inadvertently use - or for future modifications to inadvertently use.

Note that it's unsafe to interchange use of `clusterMsg` and `clusterMsgLight` in either direction as both have a `data` field which is at different offsets.

This could be made more type-safe by creating functions to perform the casting only when it's appropriate.  This would ensure creation of the properly typed variables.  Something like
```c
clusterMsg * toClusterMsg(char *buf); // asserts if light message
clusterMsgLight * toClusterMsgLight(char *buf); // asserts if full messsage
```

A new type could be used in ambiguous situations.  Static assertions can ensure that the field offsets match the two different formats.
```c
typedef struct {
    char sig[4];     /* Signature "RCmb" (Cluster message bus). */
    uint32_t totlen; /* Total length of this message */
    uint16_t ver;    /* Protocol version, currently set to CLUSTER_PROTO_VER. */
    uint16_t notused1;
    uint16_t type; /* Message type */
    uint16_t notused2;
    // "data" not here because it's at different offsets
} clusterMsgUnknown;
```

Note that the casting is only performed in 3 functions:
* clusterProcessPacket (here)
* clusterIsPacketValid - brittle
* clusterReadHandler - Only actually uses fields from light message, but calls `isClusterMsgSignatureAndLengthValid` which receives a `clusterMsg *` and then treats it ambiguously (a good use case for `clusterMsgUnknown`).

`clusterIsPacketValid` seems brittle as it pre-checks that it's a light header for known supported cases - and then trusts the code below, for various different message types, to behave appropriately.  In the case where a light message is used, the buffer is re-cast as a light message.

Code after:

    return 1;
}

/* When this function is called, there is a packet to process starting
 * at link->rcvbuf. Releasing the buffer is up to the caller, so this
 * function should just handle the higher level stuff of processing the
 * packet, modifying the cluster state if needed.
 *
 * The function returns 1 if the link is still valid after the packet
 * was processed, otherwise 0 if the link was freed since the packet
 * processing lead to some inconsistency error (for instance a PONG
 * received from the wrong sender ID). */
int clusterProcessPacket(clusterLink *link) {
    /* Validate that the packet is well-formed */
    if (!clusterIsValidPacket(link)) {
        clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
        uint16_t type = ntohs(hdr->type);
        if (server.debug_cluster_close_link_on_packet_drop &&
            (type == server.cluster_drop_packet_filter || server.cluster_drop_packet_filter == -2)) {
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %s", clusterGetMessageTypeString(type));
            return 0;
        }
        return 1;
    }

    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
    mstime_t now = mstime();
    int is_light = IS_LIGHT_MESSAGE(ntohs(hdr->type));
    uint16_t type = ntohs(hdr->type) & ~CLUSTERMSG_MODIFIER_MASK;

    if (is_light) {
        if (!link->node || nodeInHandshake(link->node)) {
            serverLog(
                LL_NOTICE,
                "Closing link for node %.40s (%s) that sent a lightweight message of type %s as its first message on the link",
                clusterLinkGetNodeName(link),
                clusterLinkGetHumanNodeName(link),
                clusterGetMessageTypeString(type));
            freeClusterLink(link);
            return 0;
        }
        clusterNode *sender = link->node;
        sender->data_received = now;
        clusterProcessLightPacket(sender, link, type);
        return 1;
    }

    uint16_t flags = ntohs(hdr->flags);
    uint64_t sender_claimed_current_epoch = 0, sender_claimed_config_epoch = 0;