diff --git a/CHANGELOG.md b/CHANGELOG.md index 669f9877..f2835276 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -143,6 +143,7 @@ Version 2.0.0 - Removed `--databaseDirectory`, `--masterPort`, and `--sslPort` command line arguments from the master server. - Implemented a new config system that replaces the old. Breaks all plugins. +- Removed usage of socket.io entirely in favor of a plain WebSocket connection. Version 1.2.4 diff --git a/client.js b/client.js index 725d185f..3d21c2dd 100644 --- a/client.js +++ b/client.js @@ -390,17 +390,19 @@ class InstanceConnection extends link.Link { } } -class SlaveConnector extends link.SocketIOClientConnector { +class SlaveConnector extends link.WebSocketClientConnector { constructor(slaveConfig) { - super(slaveConfig.get("slave.master_url"), slaveConfig.get("slave.master_token")); + super(slaveConfig.get("slave.master_url")); this.id = slaveConfig.get("slave.id"); this.name = slaveConfig.get("slave.name"); + this._token = slaveConfig.get("slave.master_token"); } register() { console.log("SOCKET | registering slave"); this.send('register_slave', { + token: this._token, agent: 'Clusterio Slave', version, id: this.id, @@ -412,8 +414,7 @@ class SlaveConnector extends link.SocketIOClientConnector { /** * Handles running the slave * - * Connects to the master server over the socket.io connection and manages - * intsances. + * Connects to the master server over the WebSocket and manages intsances. */ class Slave extends link.Link { // I don't like God classes, but the alternative of putting all this state diff --git a/clusterctl.js b/clusterctl.js index 873e6d49..edf8baef 100644 --- a/clusterctl.js +++ b/clusterctl.js @@ -1,4 +1,3 @@ -const io = require("socket.io-client"); const jwt = require("jsonwebtoken"); const fs = require("fs-extra"); const yargs = require("yargs"); @@ -315,10 +314,16 @@ commands = new Map([...commands.map(command => [command.name, command])]); /** * Connector for control connection to master server */ -class ControlConnector extends link.SocketIOClientConnector { +class ControlConnector extends link.WebSocketClientConnector { + constructor(url, token) { + super(url); + this._token = token; + } + register() { console.log("SOCKET | registering control"); this.send('register_control', { + token: this._token, agent: 'clusterctl', version: version, }); @@ -328,8 +333,7 @@ class ControlConnector extends link.SocketIOClientConnector { /** * Handles running the control * - * Connects to the master server over the socket.io connection and sends - * commands to it. + * Connects to the master server over WebSocket and sends commands to it. */ class Control extends link.Link { diff --git a/docs/socket.md b/docs/socket.md index 523cf684..790e7b0d 100644 --- a/docs/socket.md +++ b/docs/socket.md @@ -5,21 +5,9 @@ Socket.io protocol slaves, and unless you're developing Clusterio it's probably not going to be very useful for you. -The protocol is built on the [socket.io](https://socket.io/) library -which lets mulitple streams be multiplexed in many different ways over -the same connection and abstracts away reconnection logic. -Unfortunately I've been unable to figure what the message delivery -guarantees of the library is, so whether or not messages can be lost is -still an unknown, I assume that messages are delivered as long as a -connection is maintained or manages to reconnect. - -While there are many mechanisms to multiplex streams, the core -communication between the master server and slaves is done entirely by -sending single POJO payloads over the `message` event. There's a -shortcut for sending these via the `.send` method of the socket.io -sockets. - -These messages contain the following three properties: +The protocol is built on WebSocket and uses JSON encoded text messages +to communicate to and from the master server. These messages contain +the following three properties: - seq - integer - The sequence number of the message - type - string - The type of message - data - object - Message data @@ -28,7 +16,7 @@ These messages contain the following three properties: Handshake --------- -Upon connecting to the socket.io endpoint the server will respond with a +Upon connecting to the WebSocket endpoint the server will respond with a `hello` message containing the master server version as the data. The client is then expected to send a `register_slave` or `register_control` message back depending on what kind of client it is. If the register diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 93458546..6311ce2e 100644 --- a/lib/link/connectors.js +++ b/lib/link/connectors.js @@ -1,6 +1,6 @@ "use strict"; const events = require("events"); -const io = require("socket.io-client"); +const WebSocket = require("ws"); const schema = require("lib/schema"); @@ -8,12 +8,11 @@ const schema = require("lib/schema"); /** * Connector for master server clients */ -class SocketIOClientConnector extends events.EventEmitter { - constructor(url, token) { +class WebSocketClientConnector extends events.EventEmitter { + constructor(url) { super(); this._url = url; - this._token = token; this._seq = 1; this._state = "new"; @@ -33,7 +32,7 @@ class SocketIOClientConnector extends events.EventEmitter { * @returns the sequence number of the message sent */ send(type, data) { - this._socket.send({ seq: this._seq, type, data }); + this._socket.send(JSON.stringify({ seq: this._seq, type, data })); return this._seq++; } @@ -44,7 +43,7 @@ class SocketIOClientConnector extends events.EventEmitter { */ close(reason) { this.send('close', { reason }); - this.disconnect(); + this._socket.close(); } /** @@ -54,19 +53,13 @@ class SocketIOClientConnector extends events.EventEmitter { this._check(["new"]); this._state = "handshake"; - // Open socket.io connection to master - console.log(`SOCKET | connecting to ${this._url}`) let url = new URL(this._url); - url.searchParams.append('token', this._token); + url.pathname += "api/socket"; - // I really have to question the choice of having the path connected - // to be passed as separate parameter instead of the namespace. - let urlPath = url.pathname + "socket.io"; - url.pathname = ""; - this._socket = io(url.href, { - path: urlPath, - reconnectionAttempts: 2, + // Open WebSocket to master + console.log(`SOCKET | connecting to ${url}`) + this._socket = new WebSocket(url, { // For now we do not verify TLS certificates since the default setup is // to create a self-signed certificate. rejectUnauthorized: false, @@ -79,45 +72,35 @@ class SocketIOClientConnector extends events.EventEmitter { } _attachSocketHandlers() { + this._socket.on("close", (code, reason) => { + console.log(`SOCKET | Close (code: ${code}, reason: ${reason})`); + }); this._socket.on("error", err => { console.error("SOCKET | Error:", err); this._socket.close(); }); + this._socket.on("unexpected-response", (req, res) => { + let data = ""; + res.on('data', (chunk) => { data += chunk; }); + res.on('end', () => { + console.error(`SOCKET | Unexpeced Response ${res.statusCode}`); + console.error(data); + }); + }); - /*this._socket.on("connect_error", err => { - console.log("SOCKET | Connect error:", err); - });*/ - this._socket.on("connect_timeout", () => { - console.log("SOCKET | Connect timeout"); + this._socket.on("open", () => { + console.log("SOCKET | Open"); }); - this._socket.on("reconnect", attempt => { - console.log(`SOCKET | Reconnect no. ${attempt} succeeded`); + this._socket.on("ping", data => { + console.log(`SOCKET | Ping (data: ${data}`); }); - this._socket.on("reconnect_attempt", () => { - console.log("SOCKET | Attempting reconnect"); - }); - this._socket.on("reconnecting", attempt => { - console.log(`SOCKET | Attempting reconnect no. ${attempt}`); - }); - /*this._socket.on("reconnect_error", err => { - console.log("SOCKET | Reconnect error:", err); - });*/ - this._socket.on("reconnect_failed", () => { - console.log("SOCKET | Reconnecting failed"); - }); - /*this._socket.on("ping", () => { - console.log("SOCKET | Ping"); - }); - this._socket.on("pong", (ms) => { - // XXX Latency might be useful - console.log(`SOCKET | Pong latency ${ms}ms`); - });*/ - this._socket.on("connect", () => { - console.log("SOCKET | connected"); + this._socket.on("pong", data => { + console.log(`SOCKET | Pong (data: ${data}`); }); // Handle messages - this._socket.on("message", message => { + this._socket.on("message", data => { + let message = JSON.parse(data); if (this._state === "handshake") { this._processHandshake(message); } else if (this._state === "ready") { @@ -132,7 +115,7 @@ class SocketIOClientConnector extends events.EventEmitter { * Immediatly close the connection */ disconnect() { - this._socket.close(); + this._socket.terminate(); } _processHandshake(message) { @@ -155,7 +138,7 @@ class SocketIOClientConnector extends events.EventEmitter { } else if (type === 'close') { console.log(`SOCKET | received close from server: ${data.reason}`); this.emit('error', new Error(`server closed during handshake: ${data.reason}`)); - this._socket.close(); + this._socket.terminate(); } } @@ -224,6 +207,6 @@ class VirtualConnector extends events.EventEmitter { } module.exports = { - SocketIOClientConnector, + WebSocketClientConnector, VirtualConnector, }; diff --git a/lib/npmPostinstall.js b/lib/npmPostinstall.js index 73095cac..a83a3e45 100644 --- a/lib/npmPostinstall.js +++ b/lib/npmPostinstall.js @@ -87,10 +87,6 @@ let downloadList = [ url: "https://fonts.googleapis.com/css?family=Roboto", file: "static/external/css/Roboto.css", }, - { - url: "https://cdnjs.cloudflare.com/ajax/libs/socket.io/2.0.4/socket.io.js", - file: "static/external/socket.io.js", - }, ] // initiate folder creations and downloads diff --git a/lib/schema.js b/lib/schema.js index 8b49c9cb..788dd339 100644 --- a/lib/schema.js +++ b/lib/schema.js @@ -105,8 +105,9 @@ const clientHandshake = ajv.compile({ "type": { const: "register_slave" }, "data": { additionalProperties: false, - required: ["agent", "version", "name", "id"], + required: ["token", "agent", "version", "name", "id"], properties: { + "token": { type: "string" }, "agent": { type: "string" }, "version": { type: "string" }, "name": { type: "string" }, @@ -120,8 +121,9 @@ const clientHandshake = ajv.compile({ "type": { const: "register_control" }, "data": { additionalProperties: false, - required: ["agent", "version"], + required: ["token", "agent", "version"], properties: { + "token": { type: "string" }, "agent": { type: "string" }, "version": { type: "string" }, } diff --git a/master.js b/master.js index 853e5050..29e8b8cc 100644 --- a/master.js +++ b/master.js @@ -23,6 +23,7 @@ const request = require("request"); const setBlocking = require("set-blocking"); const events = require("events"); const yargs = require("yargs"); +const WebSocket = require("ws"); const version = require("./package").version; // ugly globals @@ -292,9 +293,6 @@ app.get("/api/getFactorioLocale", function(req,res){ } }); -// socket.io connection for slaves -var io = require("socket.io")({}); - class BaseConnection extends link.Link { constructor(target, connector) { super('master', target, connector); @@ -541,17 +539,9 @@ class SlaveConnection extends BaseConnection { } } -// Authentication middleware for socket.io connections -io.use((socket, next) => { - let token = socket.handshake.query.token; - authenticate.check(token).then((result) => { - if (result.ok) { - next(); - } else { - console.error(`SOCKET | authentication failed for ${socket.handshake.address}`); - next(new Error(result.msg)); - } - }); +const wss = new WebSocket.Server({ + noServer: true, + path: "/api/socket", }); /** @@ -569,7 +559,7 @@ function isInteger(value) { /** * Connector for master server connections */ -class SocketIOServerConnector extends events.EventEmitter { +class WebSocketServerConnector extends events.EventEmitter { constructor(socket) { super(); @@ -577,9 +567,9 @@ class SocketIOServerConnector extends events.EventEmitter { this._socket = socket; this._socket.on('message', message => { - this.emit('message', message); + this.emit('message', JSON.parse(message)); }); - this._socket.on('disconnect', () => { + this._socket.on('close', (code, reason) => { this.emit('disconnect'); }); } @@ -590,7 +580,7 @@ class SocketIOServerConnector extends events.EventEmitter { * @returns the sequence number of the message sent */ send(type, data = {}) { - this._socket.send({ seq: this._seq, type, data }); + this._socket.send(JSON.stringify({ seq: this._seq, type, data })); return this._seq++; } @@ -601,53 +591,73 @@ class SocketIOServerConnector extends events.EventEmitter { */ close(reason) { this.send('close', { reason }); - this.disconnect(); + this._socket.close(); } /** * Immediatly close the connection */ disconnect() { - this._socket.disconnect(true); + this._socket.terminate(); } } -io.on('connection', function (socket) { - console.log(`SOCKET | new connection from ${socket.handshake.address}`); +wss.on('connection', function (socket, req) { + console.log(`SOCKET | new connection from ${socket.remoteAddress}`); // Start connection handshake - let connector = new SocketIOServerConnector(socket); + let connector = new WebSocketServerConnector(socket); connector.send('hello', { version }); // Handle socket handshake socket.once('message', (payload) => { + payload = JSON.parse(payload); + // XXX prometheusWsUsageCounter.labels('message', "other").inc(); if (!schema.clientHandshake(payload)) { - console.log(`SOCKET | closing ${socket.handshake.address} after receiving invalid handshake`, payload); - connector.send('close', { reason: "Invalid handshake" }); - connector.disconnect(true); + console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid handshake`, payload); + connector.close("Invalid handshake"); return; } let { seq, type, data } = payload; - if (type === "register_slave") { - let connection = slaveConnections.get(data.id); - if (connection) { - console.log(`SOCKET | disconecting existing connection for slave ${data.id}`); - connection.connector.close("Registered from another connection"); - } + if (["register_slave", "register_control"].includes(type)) { + authenticate.check(data.token).then((result) => { + if (result.ok) { + if (type === "register_slave") { + let connection = slaveConnections.get(data.id); + if (connection) { + console.log(`SOCKET | disconecting existing connection for slave ${data.id}`); + connection.connector.close("Registered from another connection"); + } - console.log(`SOCKET | registered slave ${data.id} version ${data.version}`); - slaveConnections.set(data.id, new SlaveConnection(data, connector)); - connector.send('ready'); + console.log(`SOCKET | registered slave ${data.id} version ${data.version}`); + slaveConnections.set(data.id, new SlaveConnection(data, connector)); + connector.send('ready'); - } else if (type === "register_control") { - console.log(`SOCKET | registered control from ${socket.handshake.address}`); - controlConnections.push(new ControlConnection(data, connector)); - connector.send('ready'); + } else { // type === "register_control" + console.log(`SOCKET | registered control from ${req.socket.remoteAddress}`); + controlConnections.push(new ControlConnection(data, connector)); + connector.send('ready'); + } + + } else { + console.error(`SOCKET | authentication failed for ${req.socket.remoteAddress}`); + connector.close("Authentication failed"); + } + }).catch((err) => { + console.error(` ++---------------------------------------------------------------------+ +| Unexpected error occured while authenticating client, please report | +| it to https://github.com/clusterio/factorioClusterio/issues | ++---------------------------------------------------------------------+` + ); + console.error(err); + connection.close("Unexpected error"); + }); } else if (type === "close") { - console.log(`SOCKET | received close from ${socket.handshake.address}: ${data.reason}`); + console.log(`SOCKET | received close from ${req.socket.remoteAddress}: ${data.reason}`); connector.disconnect(); return; } @@ -693,10 +703,21 @@ async function loadPlugins(pluginInfos) { /** * Calls listen on server capturing any errors that occurs - * binding to the port. + * binding to the port. Also adds handler for WebSocket + * upgrade event. */ function listen(server, ...args) { return new Promise((resolve, reject) => { + server.on("upgrade", (req, socket, head) => { + console.log("handling upgrade"); + + // For reasons that defy common sense, the connection event has + // to be emitted explictly when using noServer. + wss.handleUpgrade(req, socket, head, (ws) => { + wss.emit('connection', ws, req); + }); + }); + function wrapError(err) { reject(new errors.StartupError( `Server listening failed: ${err.message}` @@ -826,7 +847,6 @@ async function startServer() { if (httpPort) { httpServer = require("http").Server(app); await listen(httpServer, httpPort); - io.attach(httpServer); console.log("Listening for HTTP on port %s...", httpServer.address().port); } @@ -847,10 +867,6 @@ async function startServer() { cert: certificate }, app) await listen(httpsServer, httpsPort); - - // XXX I'm uncertain whether or not socket.io actually supports - // attaching to multiple servers at the same time. - io.attach(httpsServer); console.log("Listening for HTTPS on port %s...", httpsServer.address().port); } } @@ -861,7 +877,7 @@ module.exports = { // For testing only _db: db, _setConfig, - _SocketIOServerConnector: SocketIOServerConnector, + _WebSocketServerConnector: WebSocketServerConnector, _controlConnections: controlConnections, _ControlConnection: ControlConnection, _slaveConnections: slaveConnections, diff --git a/package.json b/package.json index 2b258eb0..82dde234 100644 --- a/package.json +++ b/package.json @@ -61,10 +61,9 @@ "set-blocking": "^2.0.0", "simple-git": "^1.96.0", "single-line-log": "^1.1.2", - "socket.io": "^2.2.0", - "socket.io-client": "^2.2.0", "sync-request": "^4.0.1", "typescript-collections": "^1.3.2", + "ws": "^7.2.3", "yargs": "^14.2.0" }, "devDependencies": { diff --git a/test/lib/link/connector.js b/test/lib/link/connector.js index 014baa8c..ea84094a 100644 --- a/test/lib/link/connector.js +++ b/test/lib/link/connector.js @@ -6,14 +6,14 @@ const link = require('lib/link'); describe("lib/link/connectors", function() { - describe("class SocketIOClientConnector", function() { - let testConnector = new link.SocketIOClientConnector('source', 'url', 'token'); + describe("class WebSocketClientConnector", function() { + let testConnector = new link.WebSocketClientConnector('source', 'url'); testConnector._socket = new mock.MockSocket(); describe(".disconnect()", function() { - it("should call close on the socket", function() { - testConnector._socket.closeCalled = false; + it("should call terminate on the socket", function() { + testConnector._socket.terminateCalled = false; testConnector.disconnect(); - assert(testConnector._socket.closeCalled, "Close was not called"); + assert(testConnector._socket.terminateCalled, "Close was not called"); }); }); @@ -30,21 +30,22 @@ describe("lib/link/connectors", function() { it("calls send on the socket", function() { testConnector._socket.sentMessages = []; let seq = testConnector.send('test', { test: true }); - assert.deepEqual(testConnector._socket.sentMessages, [{ seq, type: 'test', data: { test: true }}]); + assert.deepEqual( + testConnector._socket.sentMessages.map(JSON.parse), + [{ seq, type: 'test', data: { test: true }}] + ); }); }); describe(".close()", function() { - it("should send close and call disconnect", function() { + it("should send close", function() { testConnector._socket.sentMessages = []; let called = false; - testConnector.disconnect = () => { called = true; } testConnector.close("test reason"); assert.deepEqual( - testConnector._socket.sentMessages, + testConnector._socket.sentMessages.map(JSON.parse), [{ seq: testConnector._seq - 1, type: 'close', data: { reason: "test reason" } }] ) - assert(called, ".disconnect() was not called"); }); }); @@ -53,7 +54,7 @@ describe("lib/link/connectors", function() { testConnector._socket.sentMessages = []; testConnector._processHandshake({ data: "invalid message" }), assert.deepEqual( - testConnector._socket.sentMessages, + testConnector._socket.sentMessages.map(JSON.parse), [{seq: testConnector._seq - 1, type: 'close', data: { reason: "Invalid handshake" } }] @@ -85,7 +86,7 @@ describe("lib/link/connectors", function() { it("should throw on message received in invalid state", function() { testConnector._state = "new"; assert.throws( - () => testConnector._socket.events.get('message')(), + () => testConnector._socket.events.get("message")("{}"), new Error("Received message in unexpected state new") ); }); @@ -93,15 +94,15 @@ describe("lib/link/connectors", function() { testConnector._state = "handshake"; let called = false; testConnector._processHandshake = () => { called = true; }; - testConnector._socket.events.get('message')(); + testConnector._socket.events.get("message")("{}"); assert(called, "_processHandshake was not called"); }); it("should emit message on message in ready state", async function() { testConnector._socket.sentMessages = []; testConnector._state = "ready"; let result = events.once(testConnector, 'message'); - testConnector._socket.events.get('message')("message"); - assert.deepEqual(await result, ["message"]); + testConnector._socket.events.get("message")('{"message":true}'); + assert.deepEqual(await result, [{ message: true }]); }); }); }); diff --git a/test/master.js b/test/master.js index 3199f58c..2506b5c0 100644 --- a/test/master.js +++ b/test/master.js @@ -63,13 +63,13 @@ describe('Master testing', function() { } }); - describe("class SocketIOServerConnector", function() { - let testConnector = new master._SocketIOServerConnector(new mock.MockSocket()); + describe("class WebSocketServerConnector", function() { + let testConnector = new master._WebSocketServerConnector(new mock.MockSocket()); describe(".disconnect()", function() { it("should call disconnect on the socket", function() { - testConnector._socket.disconnectCalled = false; + testConnector._socket.terminateCalled = false; testConnector.disconnect(); - assert(testConnector._socket.disconnectCalled, "Disconnect was not called"); + assert(testConnector._socket.terminateCalled, "Terminate was not called"); }); }); }); diff --git a/test/mock.js b/test/mock.js index 9410bd4c..f484f309 100644 --- a/test/mock.js +++ b/test/mock.js @@ -7,7 +7,6 @@ class MockSocket { constructor() { this.sentMessages = []; this.events = new Map(); - this.handshake = { address: "socket.test" }; } send(message) { @@ -18,8 +17,8 @@ class MockSocket { this.events.set(event, fn); } - disconnect() { - this.disconnectCalled = true; + terminate() { + this.terminateCalled = true; } close() {