/** * ws.js — 零依赖 WebSocket 传输层(RFC 6455 子集),供 server.js 的 /ws/hunt 使用。 * * 设计契约:docs/design/23-hunt.md §6.1(Node 原生 http upgrade + 手写握手与帧编解码, * 不引任何 npm 包): * - 握手:校验 Upgrade/Connection/Sec-WebSocket-Key/Version=13,回 101 + * Sec-WebSocket-Accept = base64(sha1(key + GUID))(node:crypto)。 * - 帧解析:客户端帧必须带 mask(不带 → close 1002);支持 opcode 0(continuation * 拼装)/1(text)/8(close,回显)/9(ping→pong)/10(pong);7/16/64 位长度全支持; * RSV 位置位或未知 opcode → 1002;控制帧分片或 >125B → 1002;二进制帧 → 1003; * 单条消息(含分片拼装)> 4KB → 1009;text 非合法 UTF-8 → 1007。 * - 服务器→客户端帧不 mask。 * - 活性:每 15s 服务器 ping,45s 无 pong 判死(触发 'close',由上层走 §6.6 断线流程)。 * - 发送背压:每连接一个待发队列;droppable(snap)新顶旧,其余不丢;队列 >200 → close 1013。 * * 导出 accept(req, socket, head) → WsConn | null。WsConn 是 EventEmitter: * 事件 'message'(string) / 'close'();方法 send(str, {droppable}) / close(code, reason); * 属性 open。房间层(hunt-room.js)只依赖这个最小接口,测试可用假连接替身。 */ 'use strict'; const crypto = require('node:crypto'); const { EventEmitter } = require('node:events'); const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; const MAX_MESSAGE_BYTES = 4096; // 单条消息上限 4KB(§6.1) const MAX_BUFFER_BYTES = 16384; // 接收缓冲兜底(消息上限 4KB,超出必为异常流量) const MAX_QUEUE = 200; // 每连接待发队列上限(§6.1 背压) const PING_INTERVAL_MS = 15000; const PONG_TIMEOUT_MS = 45000; const utf8Decoder = new TextDecoder('utf-8', { fatal: true }); /** 握手:校验升级请求并回 101;失败回 400 并销毁 socket,返回 null。 */ function accept(req, socket, head) { const key = req.headers['sec-websocket-key']; const upgrade = String(req.headers.upgrade || '').toLowerCase(); const connHdr = String(req.headers.connection || '').toLowerCase(); const version = String(req.headers['sec-websocket-version'] || ''); const ok = upgrade === 'websocket' && connHdr.split(',').map((s) => s.trim()).includes('upgrade') && typeof key === 'string' && key.length > 0 && version === '13'; if (!ok) { socket.write('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n'); socket.destroy(); return null; } const acceptKey = crypto.createHash('sha1').update(key + WS_GUID).digest('base64'); socket.write( 'HTTP/1.1 101 Switching Protocols\r\n' + 'Upgrade: websocket\r\n' + 'Connection: Upgrade\r\n' + `Sec-WebSocket-Accept: ${acceptKey}\r\n\r\n`, ); return new WsConn(socket, head); } /** 服务器→客户端帧(不 mask)。 */ function encodeFrame(opcode, payload) { const len = payload.length; let header; if (len < 126) { header = Buffer.from([0x80 | opcode, len]); } else if (len < 65536) { header = Buffer.alloc(4); header[0] = 0x80 | opcode; header[1] = 126; header.writeUInt16BE(len, 2); } else { header = Buffer.alloc(10); header[0] = 0x80 | opcode; header[1] = 127; header.writeBigUInt64BE(BigInt(len), 2); } return Buffer.concat([header, payload]); } class WsConn extends EventEmitter { constructor(socket, head) { super(); this.socket = socket; this.open = true; // 可写状态(close 帧发出即 false) this.dead = false; // 'close' 事件只发一次 this.buf = Buffer.alloc(0); this.frag = null; // 分片拼装中:{ chunks: Buffer[], size: number } this.queue = []; // 待发队列 [{ buf, droppable }] this.flushing = false; this.lastPongAt = Date.now(); socket.on('data', (chunk) => this._onData(chunk)); socket.on('error', () => this._dead()); socket.on('close', () => this._dead()); socket.on('end', () => this._dead()); this.pingTimer = setInterval(() => this._pingTick(), PING_INTERVAL_MS); if (this.pingTimer.unref) this.pingTimer.unref(); if (head && head.length) this._onData(head); } /** 应用消息发送(text 帧)。droppable=true 的待发消息被新同类顶掉(snap 专用)。 */ send(str, opts = {}) { if (!this.open) return; const buf = encodeFrame(0x1, Buffer.from(str, 'utf8')); if (opts.droppable) this.queue = this.queue.filter((q) => !q.droppable); this.queue.push({ buf, droppable: !!opts.droppable }); if (this.queue.length > MAX_QUEUE) { this.close(1013, 'send queue overflow'); return; } this._flush(); } /** 协议关闭:发 close 帧后收束 socket(50ms 内对端不回 close 也直接销毁)。 */ close(code = 1000, reason = '') { if (!this.open) return; this.open = false; const reasonBuf = Buffer.from(String(reason), 'utf8'); const payload = Buffer.alloc(2 + Math.min(reasonBuf.length, 123)); payload.writeUInt16BE(code, 0); reasonBuf.copy(payload, 2, 0, payload.length - 2); this.queue = []; try { this.socket.write(encodeFrame(0x8, payload)); } catch (_) { /* 已坏 */ } try { this.socket.end(); } catch (_) { /* 已坏 */ } this._deadTimer = setTimeout(() => this._dead(), 50); if (this._deadTimer.unref) this._deadTimer.unref(); } // ---------------- 内部 ---------------- _flush() { if (this.flushing || !this.open) return; this.flushing = true; while (this.queue.length) { const { buf } = this.queue.shift(); const ok = this.socket.write(buf); if (!ok) { this.socket.once('drain', () => { this.flushing = false; this._flush(); }); return; } } this.flushing = false; } _dead() { if (this.dead) return; this.dead = true; this.open = false; clearInterval(this.pingTimer); clearTimeout(this._deadTimer); try { this.socket.destroy(); } catch (_) { /* 忽略 */ } this.emit('close'); } /** 15s 心跳:45s 无 pong 判死(等同断线,§6.1/§6.6)。 */ _pingTick() { if (!this.open) return; if (Date.now() - this.lastPongAt > PONG_TIMEOUT_MS) { this._dead(); return; } try { this.socket.write(encodeFrame(0x9, Buffer.alloc(0))); } catch (_) { this._dead(); } } _onData(chunk) { if (!this.open) return; this.buf = Buffer.concat([this.buf, chunk]); if (this.buf.length > MAX_BUFFER_BYTES) { this._fail(1009); return; } this._parseFrames(); } _parseFrames() { while (this.open) { if (this.buf.length < 2) return; const b0 = this.buf[0]; const b1 = this.buf[1]; const fin = (b0 & 0x80) !== 0; const opcode = b0 & 0x0f; const masked = (b1 & 0x80) !== 0; let len = b1 & 0x7f; let off = 2; if (len === 126) { if (this.buf.length < 4) return; len = this.buf.readUInt16BE(2); off = 4; } else if (len === 127) { if (this.buf.length < 10) return; const big = this.buf.readBigUInt64BE(2); if (big > BigInt(MAX_MESSAGE_BYTES)) { this._fail(1009); return; } len = Number(big); off = 10; } // 协议校验(RFC 6455 + §6.1 纪律) if ((b0 & 0x70) !== 0) return this._fail(1002); // RSV 位(无扩展协商) if (!masked) return this._fail(1002); // 客户端帧必须带 mask if (![0, 1, 2, 8, 9, 10].includes(opcode)) return this._fail(1002); if (opcode === 2) return this._fail(1003); // 二进制帧不受理 if (opcode >= 8 && (!fin || len > 125)) return this._fail(1002); // 控制帧不可分片、≤125B if (opcode > 0 && len > MAX_MESSAGE_BYTES) return this._fail(1009); if (opcode === 0 && this.frag === null) return this._fail(1002); // 无分片上下文 if ((opcode === 1 || opcode === 2) && this.frag !== null) return this._fail(1002); // 分片中又起新帧 const maskOff = off; off += 4; if (this.buf.length < off + len) return; // 等更多数据 const payload = Buffer.from(this.buf.subarray(off, off + len)); const mask = this.buf.subarray(maskOff, maskOff + 4); for (let i = 0; i < payload.length; i++) payload[i] ^= mask[i & 3]; this.buf = this.buf.subarray(off + len); this._onFrame(fin, opcode, payload); } } _onFrame(fin, opcode, payload) { if (opcode === 8) { // close:回显后收束 if (this.open) { this.queue = []; try { this.socket.write(encodeFrame(0x8, payload)); } catch (_) { /* 忽略 */ } } this.open = false; try { this.socket.end(); } catch (_) { /* 忽略 */ } this._deadTimer = setTimeout(() => this._dead(), 50); if (this._deadTimer.unref) this._deadTimer.unref(); return; } if (opcode === 9) { // ping → pong(回显载荷,立即写,不走队列) try { this.socket.write(encodeFrame(0xA, payload)); } catch (_) { this._dead(); } return; } if (opcode === 10) { // pong this.lastPongAt = Date.now(); return; } // 数据帧(text / continuation) if (opcode === 1) { this.frag = { chunks: [payload], size: payload.length }; } else if (this.frag) { this.frag.chunks.push(payload); this.frag.size += payload.length; if (this.frag.size > MAX_MESSAGE_BYTES) return this._fail(1009); } if (fin && this.frag) { const full = Buffer.concat(this.frag.chunks); this.frag = null; let str; try { str = utf8Decoder.decode(full); } catch (_) { return this._fail(1007); // 非法 UTF-8 } this.emit('message', str); } } _fail(code) { this.close(code, 'protocol error'); } } module.exports = { accept, encodeFrame, WsConn, MAX_MESSAGE_BYTES, MAX_QUEUE };