// Minimal dependency-free WebSocket (RFC 6455) for the dsh-remote link. // // Why not `ws`: DSH plugins loaded through `link:` resolve imports from their own // folder, and slaves install the plugin from a tarball without node_modules. The link // only needs binary + text frames, ping/pong and close, so a small implementation keeps // the plugin installable anywhere with zero dependencies. import crypto from 'node:crypto' import { EventEmitter } from 'node:events' import http from 'node:http' import https from 'node:https' const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11' const MAX_FRAME = 64 * 1024 * 1024 export const acceptKey = (key) => crypto.createHash('sha1').update(key + GUID).digest('base64') /** One WebSocket endpoint over a raw socket (server or client role). */ export class WsConn extends EventEmitter { constructor(socket, { client = false, head } = {}) { super() this.socket = socket this.client = client this.open = true this.buf = head && head.length ? Buffer.from(head) : Buffer.alloc(0) this.frag = null socket.setNoDelay(true) socket.on('data', (d) => this._onData(d)) socket.on('close', () => this._closed(1006, 'socket closed')) // Never throw from an unhandled 'error' event: a reset peer must not crash the host. socket.on('error', (e) => { if (this.listenerCount('error')) this.emit('error', e); this._closed(1006, String(e?.message ?? e)) }) if (this.buf.length) queueMicrotask(() => this._onData(Buffer.alloc(0))) } get bufferedAmount() { return this.socket.writableLength } _closed(code, reason) { if (!this.open) return this.open = false this.emit('close', code, reason) } _onData(chunk) { if (chunk.length) this.buf = this.buf.length ? Buffer.concat([this.buf, chunk]) : chunk for (;;) { const b = this.buf if (b.length < 2) return const fin = (b[0] & 0x80) !== 0 const op = b[0] & 0x0f const masked = (b[1] & 0x80) !== 0 let len = b[1] & 0x7f let off = 2 if (len === 126) { if (b.length < 4) return len = b.readUInt16BE(2); off = 4 } else if (len === 127) { if (b.length < 10) return const big = b.readBigUInt64BE(2) if (big > BigInt(MAX_FRAME)) return this.destroy(1009, 'frame too large') len = Number(big); off = 10 } if (len > MAX_FRAME) return this.destroy(1009, 'frame too large') const maskOff = off if (masked) off += 4 if (b.length < off + len) return let payload = b.subarray(off, off + len) if (masked) { const mask = b.subarray(maskOff, maskOff + 4) payload = Buffer.from(payload) for (let i = 0; i < payload.length; i++) payload[i] ^= mask[i & 3] } this.buf = b.subarray(off + len) this._frame(fin, op, payload) if (!this.open) return } } _frame(fin, op, payload) { if (op === 0x8) { const code = payload.length >= 2 ? payload.readUInt16BE(0) : 1005 try { this._send(0x8, payload.subarray(0, 2)) } catch {} this.socket.end() return this._closed(code, payload.subarray(2).toString('utf8')) } if (op === 0x9) return this._send(0xa, payload) if (op === 0xa) return this.emit('pong', payload) if (op === 0x0) { if (!this.frag) return this.destroy(1002, 'unexpected continuation') this.frag.parts.push(payload) if (!fin) return const data = Buffer.concat(this.frag.parts) const binary = this.frag.op === 0x2 this.frag = null return this.emit('message', binary ? data : data.toString('utf8'), binary) } if (op === 0x1 || op === 0x2) { if (!fin) { this.frag = { op, parts: [payload] }; return } return this.emit('message', op === 0x2 ? Buffer.from(payload) : payload.toString('utf8'), op === 0x2) } this.destroy(1002, 'bad opcode') } _send(op, data) { if (!this.socket.writable) return false const payload = Buffer.isBuffer(data) ? data : Buffer.from(data) const len = payload.length let header if (len < 126) { header = Buffer.alloc(2); header[1] = len } else if (len < 65536) { header = Buffer.alloc(4); header[1] = 126; header.writeUInt16BE(len, 2) } else { header = Buffer.alloc(10); header[1] = 127; header.writeBigUInt64BE(BigInt(len), 2) } header[0] = 0x80 | op if (this.client) { header[1] |= 0x80 const mask = crypto.randomBytes(4) const body = Buffer.from(payload) for (let i = 0; i < body.length; i++) body[i] ^= mask[i & 3] return this.socket.write(Buffer.concat([header, mask, body])) } return this.socket.write(Buffer.concat([header, payload])) } send(data) { if (!this.open) return false return typeof data === 'string' ? this._send(0x1, Buffer.from(data, 'utf8')) : this._send(0x2, data) } ping(data = Buffer.alloc(0)) { if (this.open) this._send(0x9, data) } close(code = 1000, reason = '') { if (!this.open) return const p = Buffer.alloc(2 + Buffer.byteLength(reason)) p.writeUInt16BE(code, 0); p.write(reason, 2) try { this._send(0x8, p) } catch {} this.socket.end() setTimeout(() => this.socket.destroy(), 2000).unref() this._closed(code, reason) } destroy(code = 1006, reason = '') { this._closed(code, reason) this.socket.destroy() } } /** Complete a server-side upgrade on (req, socket, head). Returns a WsConn or null. */ export function acceptUpgrade(req, socket, head, extraHeaders = {}) { const key = req.headers['sec-websocket-key'] if (!key || String(req.headers.upgrade).toLowerCase() !== 'websocket') { socket.end('HTTP/1.1 400 Bad Request\r\nConnection: close\r\n\r\n') return null } const lines = [ 'HTTP/1.1 101 Switching Protocols', 'Upgrade: websocket', 'Connection: Upgrade', `Sec-WebSocket-Accept: ${acceptKey(key)}`, ...Object.entries(extraHeaders).map(([k, v]) => `${k}: ${v}`), ] socket.write(lines.join('\r\n') + '\r\n\r\n') return new WsConn(socket, { head }) } export function rejectUpgrade(socket, status, message = '') { const text = { 400: 'Bad Request', 401: 'Unauthorized', 403: 'Forbidden', 404: 'Not Found', 409: 'Conflict', 502: 'Bad Gateway' }[status] || 'Error' try { socket.end(`HTTP/1.1 ${status} ${text}\r\nConnection: close\r\nContent-Type: text/plain\r\nContent-Length: ${Buffer.byteLength(message)}\r\n\r\n${message}`) } catch { socket.destroy() } } /** Client connect: resolves a WsConn, rejects with {status, body} on refusal. */ export function connect(url, { headers = {}, timeoutMs = 15000, rejectUnauthorized = true } = {}) { return new Promise((resolve, reject) => { const u = new URL(url) const secure = u.protocol === 'wss:' || u.protocol === 'https:' const key = crypto.randomBytes(16).toString('base64') const req = (secure ? https : http).request({ host: u.hostname, port: u.port || (secure ? 443 : 80), path: u.pathname + u.search, method: 'GET', servername: u.hostname, rejectUnauthorized, headers: { ...headers, Host: u.host, Connection: 'Upgrade', Upgrade: 'websocket', 'Sec-WebSocket-Version': '13', 'Sec-WebSocket-Key': key, }, timeout: timeoutMs, }) req.on('timeout', () => { req.destroy(new Error(`connect timeout after ${timeoutMs} ms`)) }) req.on('error', reject) req.on('response', (res) => { let body = '' res.setEncoding('utf8') res.on('data', (d) => { if (body.length < 4096) body += d }) res.on('end', () => { const err = new Error(`upgrade refused: HTTP ${res.statusCode} ${body.trim()}`) err.status = res.statusCode err.body = body reject(err) }) }) req.on('upgrade', (res, socket, head) => { if (res.headers['sec-websocket-accept'] !== acceptKey(key)) { socket.destroy() return reject(new Error('bad Sec-WebSocket-Accept')) } socket.setTimeout(0) resolve(new WsConn(socket, { client: true, head })) }) req.end() }) }