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

Code before:
/* 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) {
            freeClusterLink(link);
            serverLog(LL_WARNING, "Closing link for matching packet type %hu", type);
            return 0;
        }
        return 1;
    }

    clusterMsg *hdr = (clusterMsg *)link->rcvbuf;
    uint16_t type = ntohs(hdr->type);
    mstime_t now = mstime();
    clusterNode *sender = getNodeFromLinkAndMsg(link, hdr);
    /* Update the last time we saw any data from this node. We
     * use this in order to avoid detecting a timeout from a node that
     * is just sending a lot of data in the cluster bus, for instance
     * because of Pub/Sub. */
    if (sender) sender->data_received = now;

    if (sender && (type == CLUSTERMSG_TYPE_PUBLISH_LIGHT || type == CLUSTERMSG_TYPE_PUBLISHSHARD_LIGHT) &&
        nodeSupportsLightMsgHdr(sender)) {
        return pubsubProcessLightPacket(link, type);
    }

    uint16_t flags = ntohs(hdr->flags);
    uint64_t senderCurrentEpoch = 0, senderConfigEpoch = 0;

    if (sender && (hdr->mflags[0] & CLUSTERMSG_FLAG0_EXT_DATA)) {
        sender->flags |= CLUSTER_NODE_EXTENSIONS_SUPPORTED;
    }

    if (sender && (!nodeSupportsLightMsgHdr(link->node)) && (flags & CLUSTER_NODE_LIGHT_HDR_SUPPORTED)) {
        sender->flags |= CLUSTER_NODE_LIGHT_HDR_SUPPORTED;
    }

    if (sender && !nodeInHandshake(sender)) {
        /* Update our currentEpoch if we see a newer epoch in the cluster. */
        senderCurrentEpoch = ntohu64(hdr->currentEpoch);

Review comment (bug):
It's not safe to call getNodeFromLinkAndMsg for a light msg, because it can access clusterMsg fields like `hdr->sender`. This happens if the sender sends a light message as its first message on the link, before we know the node ID. It's not allowed, but we want to disconnect the node, not crash ourselves.

```suggestion
    if (!link->node || nodeInHandshake(link->node)) {
        freeClusterLink(link);
        serverLog(LL_WARNING, "Closing link for node that sent a lightweight message of type %hu as its first message on the link", type);
            return 0;
    }
    custerNode *sender = link->node;    
```

Or maybe we should do this check in `clusterIsValidPacket()` instead. You can decide.

Code after:
        clusterMsgDataPublish *publish_data;
        if (is_light) {
            clusterMsgLight *hdr_light = (clusterMsgLight *)link->rcvbuf;
            publish_data = &hdr_light->data.publish.msg;
            explen = sizeof(clusterMsgLight);
        } else {
            publish_data = &hdr->data.publish.msg;
            explen = sizeof(clusterMsg);
        }
        explen -= sizeof(union clusterMsgData);
        explen +=
            sizeof(clusterMsgDataPublish) - 8 + ntohl(publish_data->channel_len) + ntohl(publish_data->message_len);
    } else if (type == CLUSTERMSG_TYPE_FAILOVER_AUTH_REQUEST || type == CLUSTERMSG_TYPE_FAILOVER_AUTH_ACK ||
               type == CLUSTERMSG_TYPE_MFSTART) {
        explen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
    } else if (type == CLUSTERMSG_TYPE_UPDATE) {
        explen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
        explen += sizeof(clusterMsgDataUpdate);
    } else if (type == CLUSTERMSG_TYPE_MODULE) {
        explen = sizeof(clusterMsg) - sizeof(union clusterMsgData);
        explen += sizeof(clusterMsgModule) - 3 + ntohl(hdr->data.module.msg.len);
    } else {
        /* We don't know this type of packet, so we assume it's well formed. */
        explen = totlen;
    }

    if (totlen != explen) {
        serverLog(LL_WARNING, "Received invalid %s packet of length %lld but expected length %lld",
                  clusterGetMessageTypeString(type), (unsigned long long)totlen, (unsigned long long)explen);
        return 0;
    }

    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) {
            freeClusterLink(link);