Host shutdown on duplicate connection (#647)

* Update no-shadow lint rule for enums

https://typescript-eslint.io/rules/no-shadow/

* Add websocket close codes enum

* Pass close code and reason to event handlers

* Shutdown host on duplicate connection

* Add connection refusal

* Fix lib/link tests

* Response to comments

* Additional Tests

* Removed missed codes in close event

.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",
 			},
 		},
 		{

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}`);

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;

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;
 		}
 

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(`

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);

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
  */

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);

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,

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")
+		);
+	});
 });