基于RFC6455的Node.js WebSocket服务器掩码分片消息数据异常排查

你的WebSocket服务器在处理大体积分片消息时出现异常，根源在于没有处理TCP数据的拆包/粘包问题，同时缺乏对部分帧的缓冲机制。以下是具体问题分析和修复方案：

核心问题分析 #

1. TCP流式传输的不确定性：TCP是无边界的流式协议，单个WebSocket帧可能被拆分成多个data事件发送，或者多个完整帧可能合并在一个data事件中。你的代码假设每个data事件对应一个完整的WebSocket帧，这在大流量场景下完全不成立。
2. 缺失部分帧缓冲逻辑：当收到的data不足以组成完整帧时，代码直接报错丢弃，没有将剩余数据缓冲起来等待后续data事件补充。
3. 分片消息状态管理不严谨：虽然定义了isInMiddleOfGetting变量，但没有跟踪初始分片的opcode，也未验证后续分片的opcode必须为0（RFC6455强制要求）。

修复后的代码 #

修改shoymayuh函数，添加帧缓冲和循环处理逻辑：

function shoymayuh(socket, cb) {
    var payloads = [];
    var currentFragmentOpcode = null; // 跟踪分片消息的初始opcode
    var frameBuffer = Buffer.alloc(0); // 缓冲不完整的帧数据
    var callback = typeof(cb) === "function" ? cb : (() => {});

    socket.on("data", buff => {
        // 将新收到的数据追加到帧缓冲
        frameBuffer = Buffer.concat([frameBuffer, buff]);

        // 循环处理缓冲中的所有完整帧
        while (frameBuffer.length >= 2) { // 至少需要2字节的帧头
            var FIN = (frameBuffer[0] >> 7) & 1;
            var opcode = frameBuffer[0] & 0b00001111;
            var isMasked = (frameBuffer[1] >> 7) & 1;
            var lengthCode = frameBuffer[1] & 0b01111111;

            // 验证客户端帧必须带掩码（RFC6455强制要求）
            if (!isMasked) {
                frameBuffer = Buffer.alloc(0); // 清空缓冲避免后续错误
                return callback({ error: "Not masked", isMasked, buff: frameBuffer });
            }

            // 计算当前帧的总长度
            var frameLength = 2; // 基础帧头长度
            var payloadLength;

            if (lengthCode < 126) {
                payloadLength = lengthCode;
            } else if (lengthCode === 126) {
                if (frameBuffer.length < 4) break; // 不够读取16位长度，等待更多数据
                payloadLength = frameBuffer.readUInt16BE(2);
                frameLength += 2;
            } else { // lengthCode === 127
                if (frameBuffer.length < 10) break; // 不够读取64位长度，等待更多数据
                payloadLength = frameBuffer.readBigUInt64BE(2);
                frameLength += 8;
            }

            // 加上掩码的4字节和负载长度
            frameLength += 4 + Number(payloadLength);

            // 如果缓冲中的数据不足以组成完整帧，跳出循环等待后续数据
            if (frameBuffer.length < frameLength) break;

            // 提取当前完整帧并更新缓冲
            var currentFrame = frameBuffer.slice(0, frameLength);
            frameBuffer = frameBuffer.slice(frameLength);

            // 处理分片消息的状态验证
            if (currentFragmentOpcode === null) {
                // 新消息的第一帧，opcode不能为0
                currentFragmentOpcode = opcode;
                if (!FIN && opcode === 0) {
                    callback({ error: "Invalid opcode for first fragment", opcode });
                    currentFragmentOpcode = null;
                    payloads = [];
                    continue;
                }
            } else {
                // 分片后续帧，opcode必须为0
                if (opcode !== 0) {
                    callback({ error: "Invalid opcode for subsequent fragment", opcode });
                    currentFragmentOpcode = null;
                    payloads = [];
                    continue;
                }
            }

            // 读取掩码并解包负载
            var maskOffset = lengthCode < 126 ? 2 : (lengthCode === 126 ? 4 : 10);
            var maskData = currentFrame.slice(maskOffset, maskOffset + 4);
            var maskedPayload = currentFrame.slice(maskOffset + 4);
            
            var unmaskedPayload = Buffer.alloc(maskedPayload.length);
            for (var i = 0; i < maskedPayload.length; i++) {
                unmaskedPayload[i] = maskedPayload[i] ^ maskData[i % 4];
            }

            payloads.push(unmaskedPayload);

            // 如果是最后一个分片，回调结果并重置状态
            if (FIN) {
                callback({
                    success: {
                        strings: payloads.map(q => q.toString()),
                        original: payloads,
                        opcode: currentFragmentOpcode
                    },
                    info: {
                        FIN,
                        opcode: currentFragmentOpcode,
                        totalPayloadLength: payloads.reduce((sum, buf) => sum + buf.length, 0)
                    }
                });
                payloads = [];
                currentFragmentOpcode = null;
            }
        }
    });

    // 处理连接关闭时的未完成分片
    socket.on("close", () => {
        if (currentFragmentOpcode !== null) {
            callback({ error: "Connection closed before fragment message completed" });
        }
    });
}

关键修复点说明 #

• 帧缓冲机制：用frameBuffer存储跨data事件的未处理数据，确保不完整的帧能被拼接完整后再处理。
• 循环处理完整帧：在每次data事件中循环处理缓冲中的所有完整帧，解决TCP粘包问题。
• 严格的分片验证：跟踪初始分片的opcode，强制后续分片的opcode为0，完全符合RFC6455规范。
• 部分帧等待逻辑：当缓冲数据不足以组成完整帧时，跳出循环等待下一次data事件补充数据，避免错误解析。

经过上述修改后，服务器就能正确处理大体积分片消息，不会再出现不符合协议格式的错误。

内容的提问来源于stack exchange，提问作者B''H Bi'ezras -- Boruch Hashem