From a2d56d4a8de2a64027d07a0ef798848264a97949 Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Wed, 17 Mar 2021 20:20:12 +0100
Subject: [PATCH] Fix crash on continuing an active connection (#369)

Fix ServerConnector reaching an invalid state if a continue operation is
performed on an active connection.  The continue would cause the
connection established logic to be runned before the connection lost
logic of the pervious connection is perfomed, resulting in a leaked
heartbeat interval and the connection lost logic being performed on the
new connection instead of the old.  Prevent this by closing both the new
and the old connection in this case.
---
 packages/lib/link/connectors.js |  3 +++
 packages/master/master.js       | 19 ++++++++++++++++++-
 test/integration/clusterio.js   | 29 ++++++++++++++++++++++++++++-
 test/integration/index.js       |  1 +
 4 files changed, 50 insertions(+), 2 deletions(-)

diff --git a/packages/lib/link/connectors.js b/packages/lib/link/connectors.js
index ef78fefd..91a6f995 100644
--- a/packages/lib/link/connectors.js
+++ b/packages/lib/link/connectors.js
@@ -75,6 +75,9 @@ class WebSocketBaseConnector extends events.EventEmitter {
 	 * Start sending heartbeats
 	 */
 	startHeartbeat() {
+		if (this._heartbeatId) {
+			throw new Error("heartbeat is already running");
+		}
 		this._lastHeartbeat = Date.now();
 		this._heartbeatId = setInterval(() => {
 			this._doHeartbeat();
diff --git a/packages/master/master.js b/packages/master/master.js
index aec2aa05..54ac0933 100755
--- a/packages/master/master.js
+++ b/packages/master/master.js
@@ -1399,7 +1399,24 @@ class WebSocketServerConnector extends libLink.WebSocketBaseConnector {
 	 * @param {number} lastSeq - The last message the client received.
 	 */
 	continue(socket, lastSeq) {
-		this._socket.terminate();
+
+		// It's possible the previous connection hasn't closed yet due to a
+		// stale connection.  If this is the case the connector will have to
+		// wait for the previous socket to close down before it can continue
+		// from a new socket, or there will be overlapping events from both.
+		if (["connected", "closing"].includes(this._state)) {
+			this._socket.close(4003, "Session Hijacked");
+
+			// Correctly waiting for the connector to close and then
+			// handling all the edge cases like multiple continues happening
+			// in parallel or connections going stale is far too difficult.
+			// Kill the connection here and let the client reconnect later
+			// when the connector is not held up by an active connection.
+			socket.close(1013, "Session Busy");
+			socket.terminate();
+			return;
+		}
+
 		this._socket = socket;
 
 		if (this._timeoutId) {
diff --git a/test/integration/clusterio.js b/test/integration/clusterio.js
index 47050662..389bd86c 100644
--- a/test/integration/clusterio.js
+++ b/test/integration/clusterio.js
@@ -2,11 +2,15 @@
 const assert = require("assert").strict;
 const fs = require("fs-extra");
 const path = require("path");
+const events = require("events");
 
 const libLink = require("@clusterio/lib/link");
 const libUsers = require("@clusterio/lib/users");
 
-const { slowTest, get, exec, execCtl, sendRcon, getControl, instancesDir } = require("./index");
+const {
+	TestControl, TestControlConnector, url, controlToken, slowTest,
+	get, exec, execCtl, sendRcon, getControl, instancesDir,
+} = require("./index");
 
 
 async function getInstances() {
@@ -32,6 +36,29 @@ describe("Integration of Clusterio", function() {
 				);
 			});
 		});
+
+		describe("run", function() {
+			it("should handle resume of an active connection", async function() {
+				slowTest(this);
+				let tlsCa = await fs.readFile("test/file/tls/cert.pem");
+				let connectorA = new TestControlConnector(url, 2, tlsCa);
+				connectorA.token = controlToken;
+				let controlA = new TestControl(connectorA);
+				await connectorA.connect();
+				connectorA._state = "closing";
+				connectorA.stopHeartbeat();
+				connectorA.on("error", () => {});
+
+				let connectorB = new TestControlConnector(url, 2, tlsCa);
+				connectorB.token = controlToken;
+				let controlB = new TestControl(connectorB);
+				connectorB._sessionToken = connectorA._sessionToken;
+				connectorB._state = "handshake";
+				await connectorB._doConnect();
+				await events.once(connectorB, "resume");
+				await connectorB.close();
+			});
+		});
 	});
 
 	describe("clusterioctl", function() {
diff --git a/test/integration/index.js b/test/integration/index.js
index be6eb761..d1588ae9 100644
--- a/test/integration/index.js
+++ b/test/integration/index.js
@@ -197,6 +197,7 @@ module.exports = {
 	getControl,
 
 	url,
+	controlToken,
 	instancesDir,
 	databaseDir,
 	masterConfigPath,