diff --git a/.eslintrc.js b/.eslintrc.js index 659219fd..a2f84e90 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -23,7 +23,9 @@ module.exports = { "@typescript-eslint/no-useless-constructor": "error", "@typescript-eslint/no-unused-vars": "off", "@typescript-eslint/triple-slash-reference": "off", + "@typescript-eslint/no-shadow": ["error", { "allow": ["Event", "Request", "yargs"] }], "prefer-const": "off", + "no-shadow": "off", }, }, { diff --git a/packages/controller/src/BaseConnection.ts b/packages/controller/src/BaseConnection.ts index 75832538..18b02086 100644 --- a/packages/controller/src/BaseConnection.ts +++ b/packages/controller/src/BaseConnection.ts @@ -41,7 +41,7 @@ export default class BaseConnection extends lib.Link { async disconnect(code: number, reason: string) { try { - await this.connector.disconnect(); + await this.connector.disconnect(code, reason); } catch (err: any) { if (!(err instanceof lib.SessionLost)) { logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`); diff --git a/packages/controller/src/HostConnection.ts b/packages/controller/src/HostConnection.ts index 7dda1d44..388014d2 100644 --- a/packages/controller/src/HostConnection.ts +++ b/packages/controller/src/HostConnection.ts @@ -152,6 +152,15 @@ export default class HostConnection extends BaseConnection { return this.info.id; } + /** + * Remote address of the host this connection is connected to + * + * @returns {string} host remote address. + */ + get remoteAddress() { + return this.info.remoteAddress; + } + async handleHostInfoUpdateEvent(event: lib.HostInfoUpdateEvent) { this.info.name = event.update.name; this.info.publicAddress = event.update.publicAddress; diff --git a/packages/controller/src/WsServer.ts b/packages/controller/src/WsServer.ts index 05576336..9def1c50 100644 --- a/packages/controller/src/WsServer.ts +++ b/packages/controller/src/WsServer.ts @@ -78,11 +78,11 @@ export default class WsServer { let disconnectTasks = []; for (let controlConnection of this.controlConnections.values()) { - disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit")); + disconnectTasks.push(controlConnection.disconnect(lib.ConnectionClosed.GoingAway, "Server Quit")); } for (let hostConnection of this.hostConnections.values()) { - disconnectTasks.push(hostConnection.disconnect(1001, "Server Quit")); + disconnectTasks.push(hostConnection.disconnect(lib.ConnectionClosed.GoingAway, "Server Quit")); } logger.info(`WsServer | Waiting for ${disconnectTasks.length} connectors to close`); @@ -97,7 +97,7 @@ export default class WsServer { } for (let socket of this.pendingSockets) { - socket.close(1001, "Server Quit"); + socket.close(lib.ConnectionClosed.GoingAway, "Server Quit"); } } @@ -170,7 +170,7 @@ export default class WsServer { ${err.stack}` ); wsRejectedConnectionsCounter.inc(); - socket.close(1011, "Unexpected error"); + socket.close(lib.ConnectionClosed.InternalError, "Unexpected error"); }); }); } @@ -184,7 +184,7 @@ ${err.stack}` } catch (err: any) { logger.verbose(`WsServer | closing ${this.remoteAddr(req)} after receiving invalid message`); wsRejectedConnectionsCounter.inc(); - socket.close(1002, `Invalid message: ${err.message}`); + socket.close(lib.ConnectionClosed.ProtocolError, `Invalid message: ${err.message}`); return; } @@ -196,14 +196,14 @@ ${err.stack}` if (message.type !== "registerHost" && message.type !== "registerControl") { logger.verbose(`WsServer | closing ${this.remoteAddr(req)} after receiving invalid handshake`); wsRejectedConnectionsCounter.inc(); - socket.close(1002, "Bad handshake"); + socket.close(lib.ConnectionClosed.ProtocolError, "Bad handshake"); return; } if (this.stopAcceptingNewSessions) { logger.verbose(`WsServer | closing ${this.remoteAddr(req)}, server is shutting down`); wsRejectedConnectionsCounter.inc(); - socket.close(1001, "Shutting down"); + socket.close(lib.ConnectionClosed.GoingAway, "Shutting down"); return; } @@ -263,15 +263,26 @@ ${err.stack}` } catch (err: any) { logger.verbose(`WsServer | authentication failed for ${this.remoteAddr(req)}`); wsRejectedConnectionsCounter.inc(); - socket.close(4003, `Authentication failed: ${err.message}`); + socket.close(lib.ConnectionClosed.Unauthorized, `Authentication failed: ${err.message}`); return; } - let [connector, sessionToken] = this.createSession(new lib.Address(lib.Address.host, data.id)); let connection = this.hostConnections.get(data.id); if (connection) { - logger.verbose(`WsServer | disconnecting existing connection for host ${data.id}`); - await connection.disconnect(1008, "Registered from another connection"); + if (connection.remoteAddress !== this.remoteAddr(req)) { + logger.verbose( + `WsServer | disallowed duplicate connection from ${this.remoteAddr(req)} for host ${data.id}` + ); + wsRejectedConnectionsCounter.inc(); + socket.close(lib.ConnectionClosed.PolicyViolation, "Host already connected from another address"); + return; + } + logger.verbose( + `WsServer | disconnecting duplicate connection from host ${data.id}` + ); + await connection.disconnect( + lib.ConnectionClosed.PolicyViolation, "Newer connection for host from same address" + ); } logger.info( @@ -284,6 +295,8 @@ ${err.stack}` ); } + let [connector, sessionToken] = this.createSession(new lib.Address(lib.Address.host, data.id)); + // Handle close before HostConnection does. This avoids a host // connection event happening between the connection breaking and // it being removed from the host connections list. @@ -326,7 +339,7 @@ ${err.stack}` } catch (err: any) { logger.verbose(`WsServer | authentication failed for ${this.remoteAddr(req)}`); wsRejectedConnectionsCounter.inc(); - socket.close(4003, `Authentication failed: ${err.message}`); + socket.close(lib.ConnectionClosed.Unauthorized, `Authentication failed: ${err.message}`); return; } diff --git a/packages/host/host.ts b/packages/host/host.ts index fa195675..b0123964 100755 --- a/packages/host/host.ts +++ b/packages/host/host.ts @@ -293,8 +293,8 @@ I version of clusterio. Expect things to break. I ` ); startHost().catch(err => { - if (err instanceof lib.AuthenticationFailed) { - logger.fatal(err.message); + if (err instanceof lib.WebSocketError) { + // Already handled by connector.on("error") within Host } else if (err instanceof lib.StartupError) { logger.fatal(` diff --git a/packages/host/src/Host.ts b/packages/host/src/Host.ts index 5fd19fb7..f6f058b2 100644 --- a/packages/host/src/Host.ts +++ b/packages/host/src/Host.ts @@ -346,13 +346,17 @@ export default class Host extends lib.Link { logger.fatal(`Unexpected error reconnecting to controller:\n${err.stack}`); return this.shutdown(); }); - } else { logger.fatal("Controller connection was unexpectedly closed"); this.shutdown(); } }); + this.connector.on("error", err => { + logger.fatal(err.message); + this.shutdown(); + }); + for (let event of ["connect", "drop", "resume", "close"] as const) { this.connector.on(event, () => { let message = new lib.ControllerConnectionEvent(event); diff --git a/packages/lib/src/errors.ts b/packages/lib/src/errors.ts index cb157c50..00071895 100644 --- a/packages/lib/src/errors.ts +++ b/packages/lib/src/errors.ts @@ -61,13 +61,34 @@ export class SessionLost extends Error { code = "SessionLost"; } +/** + * Super class for all custom websocket errors + */ +export class WebSocketError extends Error { + code = "WebSocketError"; +} + /** * Thrown when WebSocket authentication failed */ -export class AuthenticationFailed extends Error { +export class AuthenticationFailed extends WebSocketError { code = "AuthenticationFailed"; } +/** + * Thrown when WebSocket protocol is violated + */ +export class ProtocolError extends WebSocketError { + code = "ProtocolError"; +} + +/** + * Thrown when WebSocket policy is violated + */ +export class PolicyViolation extends WebSocketError { + code = "PolicyViolation"; +} + /** * Errror class for known errors occuring during startup */ diff --git a/packages/lib/src/link/connectors.ts b/packages/lib/src/link/connectors.ts index 71bfc964..9bbe2c2f 100644 --- a/packages/lib/src/link/connectors.ts +++ b/packages/lib/src/link/connectors.ts @@ -9,6 +9,30 @@ import { logger } from "../logging"; import ExponentialBackoff from "../ExponentialBackoff"; import type { Request, Event } from "./link"; +/** + * Numbered codes describing why a connection was closed + * See: https://github.com/Luka967/websocket-close-codes + * See: https://www.iana.org/assignments/websocket/websocket.xhtml#close-code-number + * See: https://www.rfc-editor.org/rfc/rfc6455.html#section-7.4 + */ +export enum ConnectionClosed { + // Codes between 1000 and 1015 are pre-assigned by IANA + // A browser client can only (on demand) send 1000 from this range + NormalClosure = 1000, // Purpose fulfilled + GoingAway = 1001, // Server quit or browser navigation + ProtocolError = 1002, // Protocol not followed + PolicyViolation = 1008, // Generic code for any endpoint policy + InternalError = 1011, // Endpoint failed to fulfil request + + // Codes after 3000 are available for frameworks + // However they should be registered to IANA + Unauthorized = 3000, + Forbidden = 3003, + Timeout = 3008, + + // Codes after 4000 are available for applications + MalformedMessage = 4000, +}; /** * Base connector for links @@ -178,13 +202,13 @@ export abstract class WebSocketBaseConnector extends BaseConnector { try { json = JSON.parse(text); } catch (err) { - this.close(4000, "Malformed JSON"); + this.close(ConnectionClosed.MalformedMessage, "Malformed JSON"); return undefined; } if (!libData.Message.validate(json)) { logger.error(`Received malformed message: ${text}`); // logger.error(JSON.stringify(libData.Message.validate.errors, null, "\t")); - this.close(4000, "Malformed message"); + this.close(ConnectionClosed.MalformedMessage, "Malformed message"); return undefined; } return libData.Message.fromJSON(json); @@ -220,7 +244,7 @@ export abstract class WebSocketBaseConnector extends BaseConnector { logger.verbose("Connector | closing after heartbeat timed out"); // eslint-disable-next-line node/no-process-env if (process.env.APP_ENV === "browser") { - this._socket!.close(4008, "Heartbeat timeout"); + this._socket!.close(ConnectionClosed.Timeout, "Heartbeat timeout"); } else { this._socket!.terminate(); } @@ -294,10 +318,12 @@ export abstract class WebSocketBaseConnector extends BaseConnector { * Gracefully disconnect * * Sets closing flag an initiates the disconnect sequence. + * @param [code=1000] WebSocket close code. + * @param [reason="Disconnect"] WebSocket close reason. */ - async disconnect() { + async disconnect(code: number = ConnectionClosed.NormalClosure, reason: string = "Disconnect") { if (this._state !== "connected") { - await this.close(1000, "Disconnect"); + await this.close(code, reason); return; } @@ -316,7 +342,7 @@ export abstract class WebSocketBaseConnector extends BaseConnector { clearTimeout(timeout); } - await this.close(1000, "Disconnect"); + await this.close(code, reason); } /** @@ -458,7 +484,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector { } if (this._socket) { - this._socket!.close(1000, "Connector closing"); + this._socket!.close(ConnectionClosed.NormalClosure, "Connector closing"); } else { this._reset(); @@ -540,10 +566,23 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector { this._socket!.onclose = (event) => { const previousState = this._state; - // Authentication failed - if (event.code === 4003) { - this.emit("error", new libErrors.AuthenticationFailed(event.reason)); - this._closing = true; + switch (event.code) { + case ConnectionClosed.Unauthorized: + // Authentication failed + this.emit("error", new libErrors.AuthenticationFailed(event.reason)); + this._closing = true; + break; + case ConnectionClosed.ProtocolError: + // Connection was closed because invalid data was sent + this.emit("error", new libErrors.ProtocolError(event.reason)); + this._closing = true; + break; + case ConnectionClosed.PolicyViolation: + // Connection was closed to prevent the server entering an invalid state + this.emit("error", new libErrors.PolicyViolation(event.reason)); + this._closing = true; + break; + default: } this._socket = null; @@ -573,7 +612,7 @@ export abstract class WebSocketClientConnector extends WebSocketBaseConnector { // Log must be sent after state change is complete let message = `Connector | Close (code: ${event.code}, reason: ${event.reason})`; - if (previousState === "connected" && event.code !== 1000) { + if (previousState === "connected" && event.code !== ConnectionClosed.NormalClosure) { logger.info(message); } else { logger.verbose(message); diff --git a/test/integration/index.js b/test/integration/index.js index 62e14af9..b9289881 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -94,9 +94,20 @@ class TestControlConnector extends lib.WebSocketClientConnector { this.sendHandshake( new lib.MessageRegisterControl( new lib.RegisterControlData( - this.token, - "clusterioctl", - "test", + this.token, "test", + ) + ) + ); + } +} + +class TestHostConnector extends lib.WebSocketClientConnector { + register() { + this.sendHandshake( + new lib.MessageRegisterHost( + new lib.RegisterHostData( + this.token, "test", + this.hostId, {}, ) ) ); @@ -347,6 +358,7 @@ process.on("exit", () => { module.exports = { TestControl, TestControlConnector, + TestHostConnector, slowTest, get, exec, diff --git a/test/integration/link.js b/test/integration/link.js index 1e7037df..73178351 100644 --- a/test/integration/link.js +++ b/test/integration/link.js @@ -4,14 +4,17 @@ const events = require("events"); const fs = require("fs-extra"); const jwt = require("jsonwebtoken"); -const { TestControlConnector, TestControl, get, exec, url } = require("./index"); +const { TestControlConnector, TestHostConnector, TestControl, get, exec, url, slowTest } = require("./index"); +const { ConnectionClosed, ProtocolError, PolicyViolation, AuthenticationFailed } = require("@clusterio/lib"); let token = jwt.sign({ aud: "user", user: "test" }, Buffer.from("TestSecretDoNotUse", "base64")); +let tokenHost = jwt.sign({ aud: "host", host: 0 }, Buffer.from("TestSecretDoNotUse", "base64")); describe("Integration of lib/link", function() { let tlsCa; let control; let controlConnector; + let hostConnector; before(async function() { tlsCa = await fs.readFile("test/file/tls/cert.pem"); }); @@ -19,9 +22,13 @@ describe("Integration of lib/link", function() { controlConnector = new TestControlConnector(url, 0.2, tlsCa); controlConnector.token = token; control = new TestControl(controlConnector, false); + hostConnector = new TestHostConnector(url, 0.2, tlsCa); + hostConnector.token = tokenHost; + hostConnector.hostId = 0; }); afterEach(async function() { await controlConnector.disconnect(); + await hostConnector.disconnect(); }); it("should emitt an error if authentication failed", async function() { @@ -35,10 +42,37 @@ describe("Integration of lib/link", function() { it("should connect with proper credentials", async function() { await controlConnector.connect(); }); + it("should emit an error if closed due to ProtocolError", async function() { + await controlConnector.connect(); + controlConnector._socket.close(ConnectionClosed.ProtocolError, "Test"); + await assert.rejects( + // event.once rejects if "error" is emitted before "close" + events.once(controlConnector, "close"), + new ProtocolError("Test") + ); + }); + it("should emit an error if closed due to PolicyError", async function() { + await controlConnector.connect(); + controlConnector._socket.close(ConnectionClosed.PolicyViolation, "Test"); + await assert.rejects( + // event.once rejects if "error" is emitted before "close" + events.once(controlConnector, "close"), + new PolicyViolation("Test") + ); + }); + it("should emit an error if closed due to Unauthorized", async function() { + await controlConnector.connect(); + controlConnector._socket.close(ConnectionClosed.Unauthorized, "Test"); + await assert.rejects( + // event.once rejects if "error" is emitted before "close" + events.once(controlConnector, "close"), + new AuthenticationFailed("Test") + ); + }); it("should reconnect on connection lost", async function() { await controlConnector.connect(); - controlConnector._socket.close(1008, "Test"); + controlConnector._socket.close(ConnectionClosed.Normal, "Test"); await events.once(controlConnector, "resume"); }); @@ -64,10 +98,51 @@ describe("Integration of lib/link", function() { it("should properly close connector if close is called during reconnect wait", async function() { await controlConnector.connect(); - controlConnector._socket.close(1008, "Test"); + controlConnector._socket.close(ConnectionClosed.Normal, "Test"); controlConnector.once("drop", () => { controlConnector.close(); }); await events.once(controlConnector, "close"); }); + + it("should refuse connection with incorrect audience", async function() { + hostConnector.token = token; + await assert.rejects( + hostConnector.connect(), + new Error("Authentication failed: jwt audience invalid. expected: host or slave") + ); + }); + + it("should refuse connection with mismatched ids", async function() { + hostConnector.hostId = 1; + await assert.rejects( + hostConnector.connect(), + new Error("Authentication failed: missmatched host id") + ); + }); + + it("should close old connections from the same address", async function() { + slowTest(this); + const hostConnectorFirst = new TestHostConnector(url, 0.2, tlsCa); + hostConnectorFirst.token = hostConnector.token; + hostConnectorFirst.hostId = hostConnector.hostId; + const onceClose = events.once(hostConnectorFirst, "close"); + await hostConnectorFirst.connect(); + await hostConnector.connect(); + await assert.rejects( + onceClose, + new PolicyViolation("Newer connection for host from same address") + ); + }); + + it("should refuse duplicate host connections", async function() { + this.skip(); // Can't be checked because the remote address for both connections is the same + + hostConnector.token = jwt.sign({ aud: "host", host: 4 }, Buffer.from("TestSecretDoNotUse", "base64")); + hostConnector.hostId = 4; // Same as already running host + await assert.rejects( + hostConnector.connect(), + new Error("Authentication failed: missmatched host id") + ); + }); });