diff --git a/packages/runtime/native/websocket-linux-arm64.node b/packages/runtime/native/websocket-linux-arm64.node new file mode 100644 index 00000000..90366234 Binary files /dev/null and b/packages/runtime/native/websocket-linux-arm64.node differ diff --git a/packages/runtime/native/websocket-linux-x64.node b/packages/runtime/native/websocket-linux-x64.node new file mode 100644 index 00000000..f336da32 Binary files /dev/null and b/packages/runtime/native/websocket-linux-x64.node differ diff --git a/packages/runtime/native/websocket-win32-x64.node b/packages/runtime/native/websocket-win32-x64.node new file mode 100644 index 00000000..b015be71 Binary files /dev/null and b/packages/runtime/native/websocket-win32-x64.node differ diff --git a/packages/websocket/client.js b/packages/websocket/client.js new file mode 100644 index 00000000..73053a0e --- /dev/null +++ b/packages/websocket/client.js @@ -0,0 +1,240 @@ +import { EventEmitter } from 'node:events'; +import net from 'node:net'; +import tls from 'node:tls'; +import crypto from 'node:crypto'; +import { URL } from 'node:url'; +import native from './native.js'; +import { + offerPerMessageDeflate, + parseAcceptedPerMessageDeflate, +} from './extensions.js'; +import { WebSocket, CONNECTING, OPEN, isValidCloseCode } from './websocket.js'; + +function parseHeaders(headerBlock) { + const lines = headerBlock.split('\r\n'); + const status = lines.shift(); + const m = /^HTTP\/1\.1\s+(\d+)\s*(.*)$/.exec(status); + if (!m) throw new Error('Bad HTTP status line: ' + status); + const headers = Object.create(null); + for (const line of lines) { + if (!line) continue; + const idx = line.indexOf(':'); + if (idx < 0) continue; + const k = line.slice(0, idx).trim().toLowerCase(); + const v = line.slice(idx + 1).trim(); + headers[k] = v; + } + return { statusCode: Number(m[1]), statusMessage: m[2], headers }; +} + +class WebSocketClient extends EventEmitter { + constructor(address, protocols, options) { + super(); + if (typeof protocols === 'object' && !Array.isArray(protocols) && protocols !== null) { + options = protocols; + protocols = undefined; + } + options = options || {}; + this._url = new URL(address); + if (this._url.protocol !== 'ws:' && this._url.protocol !== 'wss:') { + throw new Error('Only ws: and wss: URLs are supported'); + } + this._secure = this._url.protocol === 'wss:'; + this._readyState = CONNECTING; + this._maxPayload = options.maxPayload; + this._ws = null; + this._pendingQueue = []; + this._perMessageDeflate = options.perMessageDeflate; + this._protocols = Array.isArray(protocols) ? protocols.map(String) : undefined; + this.protocol = ''; + this.extensions = ''; + + const host = this._url.hostname; + const port = this._url.port ? Number(this._url.port) : (this._secure ? 443 : 80); + + // Generate random 16-byte Sec-WebSocket-Key per RFC 6455 §4.1. + const keyBytes = crypto.randomBytes(16); + this._wsKey = keyBytes.toString('base64'); + this._expectedAccept = native.computeAcceptKey(this._wsKey); + + const connectOpts = { + host, + port, + ...(options.socketOptions || {}), + }; + + const onConnect = () => { + const pathAndQuery = (this._url.pathname || '/') + (this._url.search || ''); + const hostHeader = (this._url.port && Number(this._url.port) !== (this._secure ? 443 : 80)) + ? `${host}:${this._url.port}` + : host; + const reqLines = [ + `GET ${pathAndQuery} HTTP/1.1`, + `Host: ${hostHeader}`, + 'Upgrade: websocket', + 'Connection: Upgrade', + `Sec-WebSocket-Key: ${this._wsKey}`, + 'Sec-WebSocket-Version: 13', + ]; + if (Array.isArray(protocols) && protocols.length > 0) { + reqLines.push(`Sec-WebSocket-Protocol: ${protocols.join(', ')}`); + } + const extensionOffer = offerPerMessageDeflate(options.perMessageDeflate); + if (extensionOffer) { + reqLines.push(`Sec-WebSocket-Extensions: ${extensionOffer}`); + } + if (options.headers) { + for (const [k, v] of Object.entries(options.headers)) { + reqLines.push(`${k}: ${v}`); + } + } + reqLines.push('', ''); + this._socket.write(reqLines.join('\r\n')); + }; + + if (this._secure && connectOpts.servername === undefined && net.isIP(host) === 0) { + connectOpts.servername = host; + } + + this._socket = this._secure + ? tls.connect(connectOpts, onConnect) + : net.connect(connectOpts, onConnect); + + this._socket.setNoDelay && this._socket.setNoDelay(true); + this._handshakeBuf = Buffer.alloc(0); + this._socket.on('data', (chunk) => this._onHandshakeData(chunk)); + this._socket.on('error', (err) => this._onEarlyError(err)); + this._socket.on('close', () => this._onEarlyClose()); + } + + get readyState() { + return this._ws ? this._ws.readyState : this._readyState; + } + + _onHandshakeData(chunk) { + this._handshakeBuf = Buffer.concat([this._handshakeBuf, chunk]); + const idx = this._handshakeBuf.indexOf('\r\n\r\n'); + if (idx < 0) { + if (this._handshakeBuf.length > 16384) { + this._abort('Handshake response too large'); + } + return; + } + const headBlock = this._handshakeBuf.slice(0, idx).toString('latin1'); + const rest = this._handshakeBuf.slice(idx + 4); + let parsed; + try { + parsed = parseHeaders(headBlock); + } catch (e) { + return this._abort(e.message); + } + + if (parsed.statusCode !== 101) { + return this._abort(`Unexpected response status ${parsed.statusCode}`); + } + if ((parsed.headers['upgrade'] || '').toLowerCase() !== 'websocket') { + return this._abort('Missing/invalid Upgrade header'); + } + if (!/\bupgrade\b/i.test(parsed.headers['connection'] || '')) { + return this._abort('Missing Connection: Upgrade header'); + } + if (parsed.headers['sec-websocket-accept'] !== this._expectedAccept) { + return this._abort('Invalid Sec-WebSocket-Accept'); + } + let acceptedExtensions; + try { + acceptedExtensions = parseAcceptedPerMessageDeflate( + parsed.headers['sec-websocket-extensions'], + this._perMessageDeflate, + ); + } catch (e) { + return this._abort(e.message); + } + if (parsed.headers['sec-websocket-extensions'] && !acceptedExtensions) { + return this._abort('Unexpected Sec-WebSocket-Extensions'); + } + const protocol = parsed.headers['sec-websocket-protocol'] || ''; + if (protocol && (!Array.isArray(this._protocols) || !this._protocols.includes(protocol))) { + return this._abort('Unexpected Sec-WebSocket-Protocol'); + } + + // Detach handshake listeners; hand the socket off to the WebSocket. + this._socket.removeAllListeners('data'); + this._socket.removeAllListeners('error'); + this._socket.removeAllListeners('close'); + + this._readyState = OPEN; + this._ws = new WebSocket(this._socket, { + isServer: false, + maxPayload: this._maxPayload, + extensions: acceptedExtensions ? { perMessageDeflate: acceptedExtensions } : undefined, + protocol, + readyState: OPEN, + }); + this.protocol = protocol; + this.extensions = acceptedExtensions ? 'permessage-deflate' : ''; + // Forward events to the client wrapper. + this._ws.on('message', (d, isBinary) => this.emit('message', d, isBinary)); + this._ws.on('ping', (d) => this.emit('ping', d)); + this._ws.on('pong', (d) => this.emit('pong', d)); + this._ws.on('close', (c, r) => this.emit('close', c, r)); + this._ws.on('error', (e) => this.emit('error', e)); + + this.emit('upgrade', parsed); + this.emit('open'); + + // Feed leftover bytes (if any) into the frame parser. + if (rest.length > 0) this._ws._onData(rest); + + // Flush any queued sends that happened before open. + for (const entry of this._pendingQueue) this._ws.send(entry.data, entry.options, entry.cb); + this._pendingQueue = []; + } + + _onEarlyError(err) { + this.emit('error', err); + } + + _onEarlyClose() { + if (!this._ws) { + this._readyState = 3; + this.emit('close', 1006, ''); + } + } + + _abort(msg) { + const err = new Error('WebSocket handshake failed: ' + msg); + try { this._socket.destroy(); } catch {} + this.emit('error', err); + this._readyState = 3; + this.emit('close', 1006, ''); + } + + send(data, options, cb) { + if (this._ws) return this._ws.send(data, options, cb); + this._pendingQueue.push({ data, options, cb }); + } + + ping(data, mask, cb) { if (this._ws) this._ws.ping(data, mask, cb); } + pong(data, mask, cb) { if (this._ws) this._ws.pong(data, mask, cb); } + + close(code, reason) { + if (code !== undefined && !isValidCloseCode(code)) { + throw new RangeError(`Invalid close code: ${code}`); + } + if (this._ws) return this._ws.close(code, reason); + try { this._socket.destroy(); } catch {} + } + + terminate() { + if (this._ws) return this._ws.terminate(); + try { this._socket.destroy(); } catch {} + } +} + +WebSocketClient.CONNECTING = 0; +WebSocketClient.OPEN = 1; +WebSocketClient.CLOSING = 2; +WebSocketClient.CLOSED = 3; + +export { WebSocketClient }; diff --git a/packages/websocket/extensions.js b/packages/websocket/extensions.js new file mode 100644 index 00000000..92dfa2b6 --- /dev/null +++ b/packages/websocket/extensions.js @@ -0,0 +1,152 @@ +import zlib from 'node:zlib'; + +const TRAILER = Buffer.from([0x00, 0x00, 0xff, 0xff]); + +function parseHeader(header) { + const offers = []; + if (!header) return offers; + for (const rawOffer of String(header).split(',')) { + const parts = rawOffer.split(';').map((p) => p.trim()).filter(Boolean); + if (parts.length === 0) continue; + const name = parts.shift().toLowerCase(); + const params = Object.create(null); + for (const part of parts) { + const eq = part.indexOf('='); + if (eq === -1) { + params[part.toLowerCase()] = true; + } else { + const key = part.slice(0, eq).trim().toLowerCase(); + let value = part.slice(eq + 1).trim(); + if ((value.startsWith('"') && value.endsWith('"')) || + (value.startsWith("'") && value.endsWith("'"))) { + value = value.slice(1, -1); + } + params[key] = value; + } + } + offers.push({ name, params }); + } + return offers; +} + +function normalizePerMessageDeflateOptions(value) { + if (value === undefined || value === false || value === null) return null; + if (value === true) { + return { clientNoContextTakeover: true, serverNoContextTakeover: true }; + } + if (typeof value === 'object') { + return { + clientNoContextTakeover: true, + serverNoContextTakeover: true, + ...value, + }; + } + return { clientNoContextTakeover: true, serverNoContextTakeover: true }; +} + +function acceptPerMessageDeflate(requestHeader, options) { + const config = normalizePerMessageDeflateOptions(options); + if (!config) return null; + const offer = parseHeader(requestHeader).find((item) => item.name === 'permessage-deflate'); + if (!offer) return null; + const responseParams = []; + + if (config.serverNoContextTakeover !== false || offer.params.server_no_context_takeover) { + responseParams.push('server_no_context_takeover'); + } + if (config.clientNoContextTakeover !== false || offer.params.client_no_context_takeover) { + responseParams.push('client_no_context_takeover'); + } + + const accepted = { + enabled: true, + requestNoContextTakeover: responseParams.includes('client_no_context_takeover'), + responseNoContextTakeover: responseParams.includes('server_no_context_takeover'), + threshold: config.threshold ?? 1024, + }; + + return { + header: ['permessage-deflate', ...responseParams].join('; '), + options: accepted, + }; +} + +function offerPerMessageDeflate(options) { + const config = normalizePerMessageDeflateOptions(options); + if (!config) return null; + const params = ['permessage-deflate']; + if (config.clientNoContextTakeover !== false) params.push('client_no_context_takeover'); + if (config.serverNoContextTakeover !== false) params.push('server_no_context_takeover'); + return params.join('; '); +} + +function parseAcceptedPerMessageDeflate(responseHeader, requestedOptions) { + const config = normalizePerMessageDeflateOptions(requestedOptions); + if (!config) return null; + const accepted = parseHeader(responseHeader).find((item) => item.name === 'permessage-deflate'); + if (!accepted) return null; + for (const key of Object.keys(accepted.params)) { + if (key !== 'client_no_context_takeover' && key !== 'server_no_context_takeover') { + throw new Error(`Unsupported permessage-deflate parameter: ${key}`); + } + } + return { + enabled: true, + requestNoContextTakeover: !!accepted.params.server_no_context_takeover, + responseNoContextTakeover: !!accepted.params.client_no_context_takeover, + threshold: config.threshold ?? 1024, + }; +} + +function compressRaw(data) { + const compressed = zlib.deflateRawSync(data, { flush: zlib.constants.Z_SYNC_FLUSH }); + if (compressed.length >= 4 && compressed.subarray(compressed.length - 4).equals(TRAILER)) { + return compressed.subarray(0, compressed.length - 4); + } + return compressed; +} + +function decompressRaw(data, maxPayload) { + const inflated = zlib.inflateRawSync(Buffer.concat([data, TRAILER]), { + finishFlush: zlib.constants.Z_SYNC_FLUSH, + }); + if (maxPayload !== undefined && inflated.length > maxPayload) { + const err = new Error('Message too large after inflate'); + err.code = 1009; + throw err; + } + return inflated; +} + +function normalizeProtocolList(protocols) { + if (!protocols) return []; + if (typeof protocols === 'string') return protocols.split(',').map((p) => p.trim()).filter(Boolean); + if (Array.isArray(protocols)) return protocols.map(String).map((p) => p.trim()).filter(Boolean); + if (protocols instanceof Set) return Array.from(protocols).map(String).map((p) => p.trim()).filter(Boolean); + return []; +} + +function chooseSubprotocol(requestHeader, protocols) { + const requested = normalizeProtocolList(requestHeader); + if (typeof protocols === 'function') { + const selected = protocols(requested); + return selected && requested.includes(String(selected)) ? String(selected) : null; + } + const supported = normalizeProtocolList(protocols); + if (requested.length === 0 || supported.length === 0) return null; + for (const protocol of supported) { + if (requested.includes(protocol)) return protocol; + } + return null; +} + +export { + acceptPerMessageDeflate, + chooseSubprotocol, + compressRaw, + decompressRaw, + offerPerMessageDeflate, + normalizeProtocolList, + parseAcceptedPerMessageDeflate, + parseHeader, +}; diff --git a/packages/websocket/index.js b/packages/websocket/index.js new file mode 100644 index 00000000..c1c246b1 --- /dev/null +++ b/packages/websocket/index.js @@ -0,0 +1,24 @@ +import { WebSocket as InternalWebSocket } from './websocket.js'; +import { WebSocketServer } from './server.js'; +import { WebSocketClient } from './client.js'; +import native from './native.js'; + +WebSocketClient.Server = WebSocketServer; + +const WebSocket = WebSocketClient; +const Server = WebSocketServer; +const _internal = { WebSocket: InternalWebSocket, native }; + +export { + WebSocket, + WebSocketServer, + Server, + _internal, +}; + +export default { + WebSocket, + WebSocketServer, + Server, + _internal, +}; diff --git a/packages/websocket/native.js b/packages/websocket/native.js new file mode 100644 index 00000000..ba414bfb --- /dev/null +++ b/packages/websocket/native.js @@ -0,0 +1,33 @@ +import fs from 'node:fs'; +import path from 'node:path'; +import { createRequire } from 'node:module'; +import { fileURLToPath } from 'node:url'; + +const require = createRequire(import.meta.url); +const __dirname = path.dirname(fileURLToPath(import.meta.url)); + +const candidates = [ + path.join(__dirname, '..', 'build', 'Release', 'snowluma_ws.node'), + path.join(__dirname, '..', 'build', 'Debug', 'snowluma_ws.node'), +]; + +let addon = null; +let lastErr = null; +for (const p of candidates) { + if (fs.existsSync(p)) { + try { + addon = require(p); + break; + } catch (e) { + lastErr = e; + } + } +} +if (!addon) { + throw new Error( + 'Failed to load native snowluma_ws addon. Run `npm run build` first. ' + + (lastErr ? `Last error: ${lastErr.message}` : ''), + ); +} + +export default addon; diff --git a/packages/websocket/server.js b/packages/websocket/server.js new file mode 100644 index 00000000..d24bce35 --- /dev/null +++ b/packages/websocket/server.js @@ -0,0 +1,260 @@ +import { EventEmitter } from 'node:events'; +import http from 'node:http'; +import https from 'node:https'; +import native from './native.js'; +import { + acceptPerMessageDeflate, + chooseSubprotocol, +} from './extensions.js'; +import { WebSocket, OPEN } from './websocket.js'; + +const WS_GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'; + +function computeAccept(key) { + return native.computeAcceptKey(key); +} + +// Abort the upgrade with a canned HTTP error response. +function abortUpgrade(socket, status, message) { + try { + const body = message ? message : http.STATUS_CODES[status] || ''; + const head = + `HTTP/1.1 ${status} ${http.STATUS_CODES[status] || 'Error'}\r\n` + + 'Connection: close\r\n' + + `Content-Length: ${Buffer.byteLength(body)}\r\n` + + 'Content-Type: text/plain\r\n\r\n'; + socket.write(head + body); + } catch { /* noop */ } + try { socket.destroy(); } catch { /* noop */ } +} + +function getTlsOptions(options) { + const keys = [ + 'ALPNProtocols', + 'SNICallback', + 'ca', + 'cert', + 'ciphers', + 'clientCertEngine', + 'crl', + 'dhparam', + 'ecdhCurve', + 'honorCipherOrder', + 'key', + 'maxVersion', + 'minVersion', + 'passphrase', + 'pfx', + 'privateKeyEngine', + 'privateKeyIdentifier', + 'requestCert', + 'rejectUnauthorized', + 'secureOptions', + 'secureProtocol', + 'sessionIdContext', + 'sigalgs', + 'ticketKeys', + ]; + const tlsOptions = {}; + for (const key of keys) { + if (options[key] !== undefined) tlsOptions[key] = options[key]; + } + if (options.tls && typeof options.tls === 'object') { + Object.assign(tlsOptions, options.tls); + } + return tlsOptions; +} + +class WebSocketServer extends EventEmitter { + constructor(options) { + super(); + options = options || {}; + this.options = { + port: options.port, + host: options.host, + server: options.server, + noServer: !!options.noServer, + path: options.path, + maxPayload: options.maxPayload, + verifyClient: options.verifyClient, + protocols: options.protocols, + perMessageDeflate: options.perMessageDeflate, + backlog: options.backlog, + tls: options.tls, + }; + const tlsOptions = getTlsOptions(options); + this.options.tls = Object.keys(tlsOptions).length > 0 ? tlsOptions : undefined; + this.clients = new Set(); + this._server = null; // internal http.Server (if we own it) + this._externalServer = null; + this._upgradeHandler = null; + + if (this.options.noServer) { + // Just provide handleUpgrade; user is responsible for routing. + return; + } + + if (this.options.server) { + this._externalServer = this.options.server; + this._attachToServer(this._externalServer); + } else if (this.options.port !== undefined) { + const requestHandler = (req, res) => { + // Non-WebSocket HTTP hits on the standalone port get a 426. + const body = http.STATUS_CODES[426]; + res.writeHead(426, { + 'Content-Length': Buffer.byteLength(body), + 'Content-Type': 'text/plain', + }); + res.end(body); + }; + this._server = this.options.tls + ? https.createServer(this.options.tls, requestHandler) + : http.createServer(requestHandler); + this._attachToServer(this._server); + this._server.listen( + this.options.port, + this.options.host, + this.options.backlog, + () => this.emit('listening'), + ); + this._server.on('error', (err) => this.emit('error', err)); + } else { + throw new Error('WebSocketServer requires { port } or { server } or { noServer: true }'); + } + } + + address() { + if (this._server) return this._server.address(); + if (this._externalServer && this._externalServer.address) return this._externalServer.address(); + return null; + } + + _attachToServer(server) { + const handler = (req, socket, head) => { + // If a path is configured, filter on it. + if (this.options.path) { + const urlPath = req.url.split('?')[0]; + if (urlPath !== this.options.path) { + // Let other upgrade listeners handle it; if none, abort. + if (server.listenerCount('upgrade') === 1) { + abortUpgrade(socket, 400, 'Bad path for WebSocket'); + } + return; + } + } + this.handleUpgrade(req, socket, head, (ws) => { + this.emit('connection', ws, req); + }); + }; + this._upgradeHandler = handler; + server.on('upgrade', handler); + } + + // Perform the RFC 6455 handshake on an already-produced upgrade request. + // Accepts the same argument shape as `ws` for drop-in compatibility. + handleUpgrade(request, socket, head, cb) { + const upgrade = (request.headers['upgrade'] || '').toLowerCase(); + const connection = (request.headers['connection'] || '').toLowerCase(); + const version = request.headers['sec-websocket-version']; + const key = request.headers['sec-websocket-key']; + + if (request.method !== 'GET') return abortUpgrade(socket, 405, 'Method not allowed'); + if (upgrade !== 'websocket') return abortUpgrade(socket, 400, 'Upgrade header must be websocket'); + if (!/\bupgrade\b/i.test(connection)) return abortUpgrade(socket, 400, 'Connection header must contain "upgrade"'); + if (version !== '13') { + try { + socket.write( + 'HTTP/1.1 426 Upgrade Required\r\n' + + 'Sec-WebSocket-Version: 13\r\n' + + 'Connection: close\r\n\r\n'); + } catch {} + try { socket.destroy(); } catch {} + return; + } + if (!key || !/^[+/0-9A-Za-z]{22}==$/.test(key)) { + return abortUpgrade(socket, 400, 'Invalid Sec-WebSocket-Key'); + } + + const doAccept = () => { + const accept = computeAccept(key); + const extension = acceptPerMessageDeflate( + request.headers['sec-websocket-extensions'], + this.options.perMessageDeflate, + ); + const protocol = chooseSubprotocol( + request.headers['sec-websocket-protocol'], + this.options.protocols, + ); + const responseHeaders = [ + 'HTTP/1.1 101 Switching Protocols', + 'Upgrade: websocket', + 'Connection: Upgrade', + `Sec-WebSocket-Accept: ${accept}`, + ]; + if (extension) responseHeaders.push(`Sec-WebSocket-Extensions: ${extension.header}`); + if (protocol) responseHeaders.push(`Sec-WebSocket-Protocol: ${protocol}`); + try { + socket.write(responseHeaders.join('\r\n') + '\r\n\r\n'); + } catch (err) { + try { socket.destroy(); } catch {} + return; + } + + // Disable Nagle for lower latency (typical WS guidance). + if (socket.setTimeout) socket.setTimeout(0); + if (socket.setNoDelay) socket.setNoDelay(true); + + const ws = new WebSocket(socket, { + isServer: true, + maxPayload: this.options.maxPayload, + extensions: extension ? { perMessageDeflate: extension.options } : undefined, + protocol, + readyState: OPEN, + }); + this.clients.add(ws); + ws.on('close', () => this.clients.delete(ws)); + + // If `head` contained bytes past the HTTP request (unusual for most + // clients but possible), feed them into the parser now. + if (head && head.length > 0) ws._onData(head); + + cb(ws, request); + }; + + if (typeof this.options.verifyClient === 'function') { + const info = { origin: request.headers['origin'], secure: !!(socket.encrypted), req: request }; + if (this.options.verifyClient.length >= 2) { + // Async style: (info, cb) + this.options.verifyClient(info, (allow, code, message, headers) => { + if (!allow) return abortUpgrade(socket, code || 401, message); + doAccept(); + }); + } else { + const allow = this.options.verifyClient(info); + if (!allow) return abortUpgrade(socket, 401, 'Unauthorized'); + doAccept(); + } + } else { + doAccept(); + } + } + + close(cb) { + // Stop accepting new upgrades and close our owned server (if any). + if (this._externalServer && this._upgradeHandler) { + this._externalServer.removeListener('upgrade', this._upgradeHandler); + this._upgradeHandler = null; + } + // Initiate a graceful close on all live clients. + for (const ws of this.clients) { + try { ws.close(1001, 'Server shutting down'); } catch {} + } + if (this._server) { + this._server.close((err) => cb && cb(err)); + } else if (cb) { + setImmediate(cb); + } + } +} + +export { WebSocketServer }; diff --git a/packages/websocket/websocket.js b/packages/websocket/websocket.js new file mode 100644 index 00000000..83135c21 --- /dev/null +++ b/packages/websocket/websocket.js @@ -0,0 +1,461 @@ +import { EventEmitter } from 'node:events'; +import crypto from 'node:crypto'; +import native from './native.js'; +import { compressRaw, decompressRaw } from './extensions.js'; + +// RFC 6455 opcodes. +const OP_CONT = 0x0; +const OP_TEXT = 0x1; +const OP_BIN = 0x2; +const OP_CLOSE = 0x8; +const OP_PING = 0x9; +const OP_PONG = 0xA; +const RSV1 = 0x40; + +// Ready states (mirror the standard WebSocket API numbers). +const CONNECTING = 0; +const OPEN = 1; +const CLOSING = 2; +const CLOSED = 3; + +// Default cap per-message and per-frame. +const DEFAULT_MAX_PAYLOAD = 100 * 1024 * 1024; + +function isValidCloseCode(code) { + if (code === 1000 || code === 1001 || code === 1002 || code === 1003 || + code === 1007 || code === 1008 || code === 1009 || code === 1010 || + code === 1011) { + return true; + } + if (code >= 3000 && code <= 4999) return true; + return false; +} + +function encodeClosePayload(code, reason) { + const reasonBuf = reason ? Buffer.from(String(reason), 'utf8') : Buffer.alloc(0); + if (code === undefined || code === null) { + if (reasonBuf.length > 0) { + throw new Error('Cannot send close reason without a code'); + } + return Buffer.alloc(0); + } + const buf = Buffer.allocUnsafe(2 + reasonBuf.length); + buf.writeUInt16BE(code & 0xFFFF, 0); + reasonBuf.copy(buf, 2); + return buf; +} + +// Streaming UTF-8 validator used while a text message is being received in +// fragments. Returns { ok, done } after appending a chunk; `done` means the +// message ended on a code-point boundary. +class Utf8Validator { + constructor() { + this.state = 0; // bytes still expected in current code point + this.codepoint = 0; + this.minNext = 0; // for overlong / surrogate range validation + } + + push(buf) { + for (let i = 0; i < buf.length; i++) { + const b = buf[i]; + if (this.state === 0) { + if ((b & 0x80) === 0) continue; + if ((b & 0xE0) === 0xC0) { + if (b < 0xC2) return false; // overlong + this.codepoint = b & 0x1F; + this.state = 1; + this.minNext = 0x80; + } else if ((b & 0xF0) === 0xE0) { + this.codepoint = b & 0x0F; + this.state = 2; + this.minNext = (b === 0xE0) ? 0xA0 : 0x80; + } else if ((b & 0xF8) === 0xF0) { + if (b > 0xF4) return false; + this.codepoint = b & 0x07; + this.state = 3; + this.minNext = (b === 0xF0) ? 0x90 : 0x80; + } else { + return false; + } + } else { + if (b < this.minNext || b > 0xBF) return false; + // Reject UTF-16 surrogates encoded as 3-byte sequences (ED A0..BF ..). + if (this.state === 2 && this.codepoint === 0xD && (b & 0x20)) { + // Actually this branch is unreachable because minNext would be 0x80 + // and for codepoint starting 0xED (codepoint==0xD) minNext above is + // 0x80 already so any A0..BF passes. We explicitly reject here. + return false; + } + if (this.codepoint === 0xD && this.state === 2 && b >= 0xA0) { + return false; + } + this.codepoint = (this.codepoint << 6) | (b & 0x3F); + this.state--; + this.minNext = 0x80; + } + } + return true; + } + + done() { + return this.state === 0; + } +} + +function randomMaskKey() { + const b = Buffer.allocUnsafe(4); + // crypto.randomFillSync is slightly faster than randomBytes for small sizes. + crypto.randomFillSync(b); + return b; +} + +// Shared WebSocket connection handling logic. Used by both server- and +// client-side after a successful handshake. +class WebSocket extends EventEmitter { + constructor(socket, options) { + super(); + options = options || {}; + this._socket = socket; + this._isServer = !!options.isServer; + this._maxPayload = options.maxPayload ?? DEFAULT_MAX_PAYLOAD; + this._readyState = options.readyState ?? OPEN; + this._perMessageDeflate = options.extensions && options.extensions.perMessageDeflate; + this.protocol = options.protocol || ''; + this.extensions = this._perMessageDeflate ? 'permessage-deflate' : ''; + + this._parser = new native.Parser({ + isServer: this._isServer, // server expects masked frames from clients + maxPayload: this._maxPayload, + allowedRsv: this._perMessageDeflate ? RSV1 : 0, + }); + + // Fragmented message assembly state. + this._msgOpcode = 0; + this._msgChunks = []; + this._msgSize = 0; + this._msgValidator = null; + this._msgCompressed = false; + + // Close state. + this._closeCodeSent = null; + this._closeCodeReceived = null; + this._closeReasonReceived = ''; + this._closeFrameSent = false; + this._closeTimer = null; + + this._bindSocket(); + } + + get readyState() { return this._readyState; } + get isServer() { return this._isServer; } + + _bindSocket() { + const sock = this._socket; + sock.on('data', (chunk) => this._onData(chunk)); + sock.on('end', () => this._onEnd()); + sock.on('close', () => this._onSocketClose()); + sock.on('error', (err) => this._onError(err)); + } + + _onData(chunk) { + if (this._readyState === CLOSED) return; + const res = this._parser.push(chunk); + if (res.error) { + this._failConnection(res.code || 1002, res.message || 'Protocol error'); + return; + } + for (const f of res.frames) { + this._handleFrame(f); + if (this._readyState === CLOSED) return; + } + } + + _onEnd() { + // Peer gracefully half-closed without a Close frame. + if (this._readyState !== CLOSED) { + // If we never received a Close, treat as abnormal. + if (this._closeCodeReceived === null) { + this._closeCodeReceived = 1006; + this._closeReasonReceived = ''; + } + this._destroySocket(); + } + } + + _onSocketClose() { + if (this._readyState === CLOSED) return; + if (this._closeCodeReceived === null) { + this._closeCodeReceived = 1006; + } + this._readyState = CLOSED; + if (this._closeTimer) { clearTimeout(this._closeTimer); this._closeTimer = null; } + this.emit('close', this._closeCodeReceived, this._closeReasonReceived); + } + + _onError(err) { + this.emit('error', err); + } + + _handleFrame(f) { + const { fin, opcode, payload, rsv = 0 } = f; + if (opcode === OP_PING) { + this.emit('ping', payload); + if (this._readyState === OPEN) this._sendControl(OP_PONG, payload); + return; + } + if (opcode === OP_PONG) { + this.emit('pong', payload); + return; + } + if (opcode === OP_CLOSE) { + this._handleCloseFrame(payload); + return; + } + + // Data frames (TEXT / BIN / CONT). + if (opcode === OP_TEXT || opcode === OP_BIN) { + if (this._msgOpcode !== 0) { + this._failConnection(1002, 'New data frame started before previous fragmented message finished'); + return; + } + if ((rsv & RSV1) && !this._perMessageDeflate) { + this._failConnection(1002, 'Compressed frame without negotiated permessage-deflate'); + return; + } + this._msgOpcode = opcode; + this._msgChunks = []; + this._msgSize = 0; + this._msgCompressed = (rsv & RSV1) !== 0; + if (opcode === OP_TEXT) this._msgValidator = new Utf8Validator(); + else this._msgValidator = null; + } else if (opcode === OP_CONT) { + if (this._msgOpcode === 0) { + this._failConnection(1002, 'Continuation frame without active message'); + return; + } + if (rsv !== 0) { + this._failConnection(1002, 'RSV bits set on continuation frame'); + return; + } + } + + this._msgSize += payload.length; + if (this._msgSize > this._maxPayload) { + this._failConnection(1009, 'Message too large'); + return; + } + if (payload.length > 0) this._msgChunks.push(payload); + + if (fin) { + let data = this._msgChunks.length === 1 + ? this._msgChunks[0] + : Buffer.concat(this._msgChunks, this._msgSize); + if (this._msgCompressed) { + try { + data = decompressRaw(data, this._maxPayload); + } catch (err) { + this._failConnection(err.code || 1007, err.message || 'Invalid compressed payload'); + return; + } + } + if (this._msgValidator) { + if (!this._msgValidator.push(data) || !this._msgValidator.done()) { + this._failConnection(1007, 'Invalid UTF-8'); + return; + } + } + const isBinary = this._msgOpcode === OP_BIN; + this._msgOpcode = 0; + this._msgChunks = []; + this._msgSize = 0; + this._msgValidator = null; + this._msgCompressed = false; + this.emit('message', data, isBinary); + } + } + + _handleCloseFrame(payload) { + let code = 1005; + let reason = ''; + if (payload.length === 1) { + return this._failConnection(1002, 'Close payload length 1'); + } + if (payload.length >= 2) { + code = payload.readUInt16BE(0); + if (!isValidCloseCode(code)) { + return this._failConnection(1002, 'Invalid close code'); + } + if (payload.length > 2) { + try { + reason = new TextDecoder('utf-8', { fatal: true }).decode(payload.subarray(2)); + } catch { + return this._failConnection(1007, 'Invalid UTF-8 in close reason'); + } + } + } + this._closeCodeReceived = code; + this._closeReasonReceived = reason; + + if (this._readyState === OPEN) { + // Echo a close frame back with same code (per RFC 6455 §5.5.1). + this._readyState = CLOSING; + const echo = encodeClosePayload(code === 1005 ? undefined : code, ''); + this._sendControl(OP_CLOSE, echo); + this._closeFrameSent = true; + this._endSocketSoon(); + } else if (this._readyState === CLOSING) { + // We already sent our close frame; the handshake is complete. + this._endSocketSoon(); + } + } + + _endSocketSoon() { + // Flush pending writes then half-close. Give the peer a moment to close. + try { this._socket.end(); } catch { /* noop */ } + if (this._closeTimer) clearTimeout(this._closeTimer); + this._closeTimer = setTimeout(() => this._destroySocket(), 30000); + this._closeTimer.unref && this._closeTimer.unref(); + } + + _destroySocket() { + try { this._socket.destroy(); } catch { /* noop */ } + } + + _failConnection(code, reason) { + if (this._readyState === CLOSED) return; + const payload = encodeClosePayload(code, reason); + let frame = null; + try { + frame = native.buildFrame(OP_CLOSE, true, payload, + this._isServer ? null : randomMaskKey(), 0); + } catch { /* noop */ } + this._closeCodeSent = code; + this._readyState = CLOSING; + const err = new Error(`WebSocket protocol error: ${reason}`); + err.code = code; + this.emit('error', err); + // Use `end(frame)` so the close frame flushes and a FIN is sent. This + // yields a clean TCP half-close peers can observe via the 'end' event. + try { + if (frame) this._socket.end(frame); + else this._socket.end(); + } catch { /* noop */ } + // Failsafe: destroy if the peer never FIN's back. + if (this._closeTimer) clearTimeout(this._closeTimer); + this._closeTimer = setTimeout(() => this._destroySocket(), 2000); + this._closeTimer.unref && this._closeTimer.unref(); + } + + _sendControl(opcode, payload) { + if (!payload) payload = Buffer.alloc(0); + const frame = native.buildFrame(opcode, true, payload, this._isServer ? null : randomMaskKey(), 0); + this._socket.write(frame); + } + + // Public API ------------------------------------------------------------- + + send(data, options, cb) { + if (typeof options === 'function') { cb = options; options = undefined; } + options = options || {}; + if (this._readyState !== OPEN) { + const err = new Error('WebSocket is not open'); + if (cb) return cb(err); + throw err; + } + let payload; + let opcode; + if (typeof data === 'string') { + payload = Buffer.from(data, 'utf8'); + opcode = options.binary ? OP_BIN : OP_TEXT; + } else if (Buffer.isBuffer(data)) { + payload = data; + opcode = options.binary === false ? OP_TEXT : OP_BIN; + } else if (data instanceof Uint8Array) { + payload = Buffer.from(data.buffer, data.byteOffset, data.byteLength); + opcode = options.binary === false ? OP_TEXT : OP_BIN; + } else if (data instanceof ArrayBuffer) { + payload = Buffer.from(data); + opcode = options.binary === false ? OP_TEXT : OP_BIN; + } else { + const err = new TypeError('Unsupported data type for send()'); + if (cb) return cb(err); + throw err; + } + let rsv = 0; + if (this._perMessageDeflate && options.compress !== false && + payload.length >= (this._perMessageDeflate.threshold ?? 1024)) { + payload = compressRaw(payload); + rsv = RSV1; + } + const maskKey = this._isServer ? null : randomMaskKey(); + const frame = native.buildFrame(opcode, true, payload, maskKey, rsv); + this._socket.write(frame, cb); + } + + ping(data, mask, cb) { + if (typeof data === 'function') { cb = data; data = undefined; mask = undefined; } + if (typeof mask === 'function') { cb = mask; mask = undefined; } + const payload = data ? (Buffer.isBuffer(data) ? data : Buffer.from(String(data), 'utf8')) : Buffer.alloc(0); + if (payload.length > 125) throw new Error('Ping payload must be <=125 bytes'); + if (this._readyState !== OPEN) { + const err = new Error('WebSocket is not open'); + if (cb) return cb(err); + throw err; + } + const maskKey = this._isServer ? null : randomMaskKey(); + const frame = native.buildFrame(OP_PING, true, payload, maskKey, 0); + this._socket.write(frame, cb); + } + + pong(data, mask, cb) { + if (typeof data === 'function') { cb = data; data = undefined; mask = undefined; } + if (typeof mask === 'function') { cb = mask; mask = undefined; } + const payload = data ? (Buffer.isBuffer(data) ? data : Buffer.from(String(data), 'utf8')) : Buffer.alloc(0); + if (payload.length > 125) throw new Error('Pong payload must be <=125 bytes'); + if (this._readyState !== OPEN) { + const err = new Error('WebSocket is not open'); + if (cb) return cb(err); + throw err; + } + const maskKey = this._isServer ? null : randomMaskKey(); + const frame = native.buildFrame(OP_PONG, true, payload, maskKey, 0); + this._socket.write(frame, cb); + } + + close(code, reason) { + if (this._readyState === CLOSED || this._readyState === CLOSING) return; + if (code !== undefined && !isValidCloseCode(code)) { + throw new RangeError(`Invalid close code: ${code}`); + } + const reasonBuf = reason ? Buffer.from(String(reason), 'utf8') : Buffer.alloc(0); + if (reasonBuf.length > 123) throw new RangeError('Close reason must be <=123 bytes'); + const payload = encodeClosePayload(code, reason); + const maskKey = this._isServer ? null : randomMaskKey(); + const frame = native.buildFrame(OP_CLOSE, true, payload, maskKey, 0); + this._readyState = CLOSING; + this._closeCodeSent = code ?? null; + this._socket.write(frame); + this._closeFrameSent = true; + this._endSocketSoon(); + } + + terminate() { + this._readyState = CLOSING; + this._destroySocket(); + } +} + +WebSocket.CONNECTING = CONNECTING; +WebSocket.OPEN = OPEN; +WebSocket.CLOSING = CLOSING; +WebSocket.CLOSED = CLOSED; + +export { + WebSocket, + CONNECTING, OPEN, CLOSING, CLOSED, + OP_CONT, OP_TEXT, OP_BIN, OP_CLOSE, OP_PING, OP_PONG, + RSV1, + isValidCloseCode, + encodeClosePayload, + randomMaskKey, +};