From 6c5531cf0fea2b7b6b875d905f1a053766484a58 Mon Sep 17 00:00:00 2001 From: =?utf8?q?Rapha=C3=ABl=20Jakse?= Date: Sun, 12 Apr 2020 20:14:39 +0200 Subject: [PATCH] Support for WebSockets in the server --- trivabble-server.js | 393 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 351 insertions(+), 42 deletions(-) diff --git a/trivabble-server.js b/trivabble-server.js index 814447f..a97c87b 100644 --- a/trivabble-server.js +++ b/trivabble-server.js @@ -46,6 +46,11 @@ const debuglog = DEBUG_LOG ? console.log.bind(console) : () => null; const http = require("http"); const fs = require("fs"); +const crypto = require("crypto"); + +const REQUEST_TYPE_LONG_POLLING = 1; +const REQUEST_TYPE_SSE = 2; +const REQUEST_TYPE_WEBSOCKET = 3; const frBag = [ " ", " ", // Jokers @@ -136,17 +141,25 @@ function Game() { this.pendingEvents = []; } -function writeResponse(responseAndIsES, data, terminate) { - responseAndIsES[0][terminate ? "end" : "write"]( - responseAndIsES[1] - ? "data:" + data + "\n\n" - : data.length + data - ); +function writeMessage(responseAndType, data, terminate) { + if (responseAndType[1] === REQUEST_TYPE_WEBSOCKET) { + webSocketWrite(responseAndType[0], data); + } else { + responseAndType[0][terminate ? "end" : "write"]( + responseAndType[1] === REQUEST_TYPE_SSE + ? "data:" + data + "\n\n" + : data.length + data + ); + } if (terminate) { - stopKeepAlive(responseAndIsES); + stopKeepAlive(responseAndType); + if (responseAndType[1] === REQUEST_TYPE_WEBSOCKET && responseAndType[0] && !responseAndType[0].isDestroyed) { + responseAndType[0].end(webSocketCloseBuffer); + responseAndType[0] = null; + } } else { - keepAlive(responseAndIsES); + keepAlive(responseAndType); } } @@ -167,7 +180,7 @@ function stop() { for (const gameID of Object.keys(games)) { const game = games[gameID]; for (const listeningPlayer of game.listeningPlayers) { - writeResponse(listeningPlayer, '{"stopping": 2000}', true); + writeMessage(listeningPlayer, '{"stopping": 2000}', true); } if (game.listeningPlayers.length) { @@ -181,20 +194,33 @@ function stop() { }); } -function stopKeepAlive(responseAndIsES) { - if (responseAndIsES.keepAliveTimeout) { - clearTimeout(responseAndIsES.keepAliveTimeout); - responseAndIsES.keepAliveTimeout = 0; +const webSocketPingBuffer = Buffer.from([ + 0b10001001, 0b00000000 +]); + +const webSocketCloseBuffer = Buffer.from([ + 0b10001000, 0b00000000 +]); + + +function stopKeepAlive(responseAndType) { + if (responseAndType.keepAliveTimeout) { + clearTimeout(responseAndType.keepAliveTimeout); + responseAndType.keepAliveTimeout = 0; } } -function keepAlive(responseAndIsES) { - stopKeepAlive(responseAndIsES); - responseAndIsES.keepAliveTimeout = setTimeout(function keepAliveTimeout() { - responseAndIsES[0].write( - responseAndIsES[1] - ? ":\n\n" - : "2[]" +function keepAlive(responseAndType) { + stopKeepAlive(responseAndType); + responseAndType.keepAliveTimeout = setTimeout(function keepAliveTimeout() { + responseAndType[0].write( + responseAndType[1] === REQUEST_TYPE_WEBSOCKET + ? webSocketPingBuffer + : ( + responseAndType[1] === REQUEST_TYPE_SSE + ? ":\n\n" + : "2[]" + ) ); }, KEEP_ALIVE); } @@ -288,11 +314,11 @@ Game.prototype.playerJoined = function (playerName) { this.pendingEvents.push({players: players}); }; -Game.prototype.addListeningPlayer = function (playerName, responseAndIsES) { +Game.prototype.addListeningPlayer = function (playerName, responseAndType) { const that = this; - that.listeningPlayers.push(responseAndIsES); - keepAlive(responseAndIsES); + that.listeningPlayers.push(responseAndType); + keepAlive(responseAndType); let closed = false; @@ -302,18 +328,19 @@ Game.prototype.addListeningPlayer = function (playerName, responseAndIsES) { } closed = true; - stopKeepAlive(responseAndIsES); + stopKeepAlive(responseAndType); - const index = that.listeningPlayers.indexOf(responseAndIsES); + const index = that.listeningPlayers.indexOf(responseAndType); if (index !== -1) { that.listeningPlayers[index] = that.listeningPlayers[that.listeningPlayers.length - 1]; that.listeningPlayers.pop(); } } - responseAndIsES[0].on("close", close); - responseAndIsES[0].on("finish", close); - responseAndIsES[0].on("prefinish", close); + responseAndType[0].on("error", close); + responseAndType[0].on("close", close); + responseAndType[0].on("finish", close); + responseAndType[0].on("prefinish", close); this.playerJoined(playerName); this.commit(); @@ -330,7 +357,7 @@ Game.prototype.commit = function () { } if (this.listeningPlayers[i]) { - writeResponse(this.listeningPlayers[i], msg); + writeMessage(this.listeningPlayers[i], msg); } } @@ -662,11 +689,11 @@ function handleCommand(cmd, gameNumber, playerName, response) { } } -function handleCommands(cmds, responseAndIsES) { +function handleCommands(cmds, responseAndType) { if (!cmds.cmds || !cmds.cmds.length) { const {gameNumber, game} = joinGame(cmds.gameNumber); - writeResponse(responseAndIsES, + writeMessage(responseAndType, JSON.stringify({ "playerName": cmds.playerName, "gameNumber": gameNumber, @@ -677,11 +704,28 @@ function handleCommands(cmds, responseAndIsES) { }) ); - game.addListeningPlayer(cmds.playerName, responseAndIsES); + game.addListeningPlayer(cmds.playerName, responseAndType); return; } - const response = responseAndIsES[0]; + let message = ""; + + const response = ( + responseAndType[1] === REQUEST_TYPE_WEBSOCKET + ? { + write: function (s) { + message += s; + }, + + end: function (s) { + if (s) { + message += s; + } + webSocketWrite(responseAndType[0], message); + } + } + : responseAndType[0] + ); response.write("["); @@ -695,22 +739,162 @@ function handleCommands(cmds, responseAndIsES) { response.end("]"); if (games[cmds.gameNumber]) { - debuglog("pendingEvents COMMIT", games[cmds.gameNumber].pendingEvents); games[cmds.gameNumber].commit(); } } +// Thx https://medium.com/hackernoon/implementing-a-websocket-server-with-node-js-d9b78ec5ffa8 + +function webSocketWrite(socket, data) { + /* eslint-disable capitalized-comments */ + + // Copy the data into a buffer + data = Buffer.from(data); + const byteLength = data.length; + + // Note: we're not supporting > 65535 byte payloads at this stage + const lengthByteCount = byteLength < 126 ? 0 : 2; + const payloadLength = lengthByteCount === 0 ? byteLength : 126; + const buffer = Buffer.alloc(2 + lengthByteCount + byteLength); + + // Write out the first byte, using opcode `1` to indicate that the message + // payload contains text data + buffer.writeUInt8(0b10000001, 0); + buffer.writeUInt8(payloadLength, 1); + + // Write the length of the payload to the second byte + + if (lengthByteCount === 2) { + buffer.writeUInt16BE(byteLength, 2); + } + + // Write the data to the data buffer + data.copy(buffer, 2 + lengthByteCount); + socket.write(buffer); +} + +function webSocketGetMessage(buffer) { + /* eslint-disable no-bitwise, capitalized-comments */ + + let dataAfter = ""; + let finalOffset = -1; + + if (buffer.length < 2) { + return ["", 0]; + } + + const firstByte = buffer.readUInt8(0); + + + const isFinalFrame = Boolean((firstByte >>> 7) & 0x1); + // const [reserved1, reserved2, reserved3] = [ + // Boolean((firstByte >>> 6) & 0x1), + // Boolean((firstByte >>> 5) & 0x1), + // Boolean((firstByte >>> 4) & 0x1) + // ]; + + const opCode = firstByte & 0xF; + + // We can return null to signify that this is a connection termination frame + if (opCode === 0x8) { + return null; + } + + const secondByte = buffer.readUInt8(1); + const isMasked = (secondByte >>> 7) & 0x1; + + // Keep track of our current position as we advance through the buffer + let currentOffset = 2; + let payloadLength = secondByte & 0x7F; + if (payloadLength > 125) { + if (payloadLength === 126) { + payloadLength = buffer.readUInt16BE(currentOffset); + currentOffset += 2; + } else { + // 127 + // If this has a value, the frame size is ridiculously huge! + // const leftPart = buffer.readUInt32BE(currentOffset); + // const rightPart = buffer.readUInt32BE(currentOffset += 4); + // Honestly, if the frame length requires 64 bits, you're probably doing it wrong. + // In Node.js you'll require the BigInt type, or a special library to handle this. + throw new Error("Large websocket payloads not currently implemented"); + } + } + + if (payloadLength + currentOffset > buffer.length) { + return ["", 0]; + } + + if (!isFinalFrame) { + const message = webSocketGetMessage(buffer.slice(payloadLength + currentOffset)); + if (message) { + if (message[1]) { + dataAfter = message[0]; + finalOffset = message[1]; + } else { + return ["", 0]; + } + } else if (message === null) { + // ??!? + return null; + } + } + + let maskingKey; + if (isMasked) { + maskingKey = buffer.readUInt32BE(currentOffset); + currentOffset += 4; + } + + // Allocate somewhere to store the final message data + const data = Buffer.alloc(payloadLength); + + // Only unmask the data if the masking bit was set to 1 + if (isMasked) { + // Loop through the source buffer one byte at a time, keeping track of which + // byte in the masking key to use in the next XOR calculation + for (let i = 0, j = 0; i < payloadLength; ++i, j = i % 4) { + // Extract the correct byte mask from the masking key + const shift = (j === 3) ? 0 : (3 - j) << 3; + const mask = (shift === 0 ? maskingKey : (maskingKey >>> shift)) & 0xFF; + + // Read a byte from the source buffer + const source = buffer.readUInt8(currentOffset++); + + // XOR the source byte and write the result to the data buffer + data.writeUInt8(mask ^ source, i); + } + } else { + // Not masked - we can just read the data as-is + buffer.copy(data, 0, currentOffset); + } + + return [ + opCode === 0x1 + ? (data.toString("utf8") + dataAfter) + : "", + finalOffset === -1 + ? currentOffset + : finalOffset + ]; +} + function handleRequest(request, response) { let post = ""; - const responseAndIsES = [response, false]; + const responseAndType = [response, REQUEST_TYPE_LONG_POLLING]; + let upgradedToWebSocket = false; response.on("error", function connectionError(e) { console.error("An error occurred while trying to write on a socket", e); - stopKeepAlive(responseAndIsES); + stopKeepAlive(responseAndType); }); // Thx http://stackoverflow.com/questions/4295782/how-do-you-extract-post-data-in-node-js request.on("data", function (data) { + if (upgradedToWebSocket) { + return; + } + post += data; // Too much POST data, kill the connection! @@ -719,7 +903,19 @@ function handleRequest(request, response) { } }); + request.on("upgrade", function () { + upgradedToWebSocket = true; + }); + + request.on("error", function (error) { + console.error("Error while handling this request", request, error); + }); + request.on("end", function () { + if (upgradedToWebSocket) { + return; + } + if (!running) { response.statusCode = 503; response.statusMessage = "Server is stopping"; @@ -777,18 +973,18 @@ function handleRequest(request, response) { return; } - const eventSourceMatch = request.url.match(/\/:[^/]+\/sse\/(?[\s\S]+)$/u); + const sseMatch = request.url.match(/\/:[^/]+\/sse\/(?[\s\S]+)$/u); - if (eventSourceMatch) { + if (sseMatch) { response.setHeader("Content-Type", "text/event-stream"); - post = decodeURIComponent(eventSourceMatch.groups.data); - responseAndIsES[1] = true; + post = decodeURIComponent(sseMatch.groups.data); + responseAndType[1] = REQUEST_TYPE_SSE; } else { response.setHeader("Content-Type", "text/plain"); response.setHeader("Transfer-Encoding", "chunked"); } - debuglog("RECEIVED", request.url, post, responseAndIsES[1]); + debuglog("REQ:", request.url, post, responseAndType[1]); try { post = post && JSON.parse(post); @@ -799,7 +995,7 @@ function handleRequest(request, response) { return; } - handleCommands(post, responseAndIsES); + handleCommands(post, responseAndType); }); } @@ -821,6 +1017,119 @@ fs.readFile("games.backup.json", function (err, data) { console.error("WARNING: ", e); } + // See https://developer.mozilla.org/en-US/docs/Web/API/WebSockets_API/Writing_WebSocket_servers + server.on("upgrade", function upgradeToWebsocket(request, socket) { + socket.on("error", function (e) { + console.error("Socket error", e); + }); + + // socket.on("close", function () {}); + + const webSocketMatch = request.url.match(/\/:[^/]+\/ws\/(?[\s\S]+)$/u); + + let decodedData; + try { + decodedData = decodeURIComponent(webSocketMatch.groups.data); + } catch (e) { + socket.end("HTTP/1.1 400 Bad Request"); + console.error("WS: FAILED - Could not decode the given data " + webSocketMatch.groups.data); + return; + } + + let cmds; + + if (webSocketMatch) { + try { + cmds = JSON.parse(decodedData); + } catch (e) { + socket.end("HTTP/1.1 400 Bad Request"); + console.error("WS: FAILED - Could not parse the given data " + decodedData); + return; + } + } else { + socket.end("HTTP/1.1 400 Bad Request"); + return; + } + + if (request.headers.upgrade !== "websocket") { + socket.end("HTTP/1.1 400 Bad Request"); + console.error("WS: FAILED - HTTP header 'Upgrade' is not 'websocket' " + decodedData); + return; + } + + socket.write( + "HTTP/1.1 101 Web Socket Protocol Handshake\r\n" + + "Upgrade: WebSocket\r\n" + + "Connection: Upgrade\r\n" + ); + + const webSocketKey = request.headers["sec-websocket-key"]; + + if (webSocketKey) { + socket.write( + "Sec-WebSocket-Accept: " + ( + crypto + .createHash("sha1") + .update(webSocketKey + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11", "binary") + .digest("base64") + ) + "\r\n" + ); + } + + socket.write("\r\n"); + + debuglog("WS: established " + decodedData); + + const responseAndType = [socket, REQUEST_TYPE_WEBSOCKET]; + + let receivedBytes = null; + + socket.on("data", function webSocketData(buffer) { + if (receivedBytes) { + buffer = Buffer.concat([receivedBytes, buffer]); + receivedBytes = null; + } + + while (true) { + const message = webSocketGetMessage(buffer); + if (message && message[1]) { + if (message[0]) { + debuglog("WS message: " + message[0]); + try { + cmds.cmds = JSON.parse(message[0]); + } catch (e) { + writeMessage(responseAndType, '{"error":1, "reason": "Unable to parse your JSON data"}'); + return; + } + + handleCommands(cmds, responseAndType); + cmds.cmds = null; + } + } else if (message === null) { + if (responseAndType[0] && !socket.isDestroyed) { + responseAndType[0] = null; + socket.end(webSocketCloseBuffer); + } + + return; + } + + if (message[1] === buffer.length) { + break; + } + + if (message[1] === 0) { + receivedBytes = buffer; + return; + } + + buffer = buffer.slice(message[1]); + } + }); + + handleCommands(cmds, responseAndType); + }); + server.on("error", function (error) { console.error("An error happened in the HTTP server", error); }); -- 2.30.2