factorio-clusterio/test/integration/link.js

import assert from "node:assert/strict";
import events from "node:events";
import jwt from "jsonwebtoken";

import {
	TestControlConnector, TestHostConnector, TestControl, url, controlToken, getControl, hostId, pairTestHost,
} from "./index.js";
import * as lib from "@clusterio/lib";
import { ConnectionClosed, ProtocolError, PolicyViolation, AuthenticationFailed } from "@clusterio/lib";
import { slowTest } from "../common.js";

// Valid token for a host id that was never approved
let tokenUnregistered = jwt.sign({ aud: "host", host: 0 }, Buffer.from("TestSecretDoNotUse", "base64"));

describe("Integration of lib/link", function() {
	let control;
	let controlConnector;
	let hostConnector;
	// Hosts can only connect once approved, so the host connections use a paired host
	let linkHost;
	before(async function() {
		linkHost = await pairTestHost("link-test");
	});
	after(async function() {
		if (linkHost) {
			await getControl().send(new lib.HostRevokeRequest(linkHost.hostId));
		}
	});
	beforeEach(async function() {
		controlConnector = new TestControlConnector(url, 0.2);
		controlConnector.token = controlToken;
		control = new TestControl(controlConnector, false);
		hostConnector = new TestHostConnector(url, 0.2);
		hostConnector.token = linkHost.token;
		hostConnector.hostId = linkHost.hostId;
	});
	afterEach(async function() {
		await controlConnector.disconnect();
		await hostConnector.disconnect();
	});

	it("should emitt an error if authentication failed", async function() {
		controlConnector.token = "gibberish";
		await assert.rejects(
			controlConnector.connect(),
			new Error("Authentication failed: jwt malformed")
		);
	});

	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(ConnectionClosed.Normal, "Test");
		await events.once(controlConnector, "resume");
	});

	it("should reconnect on connection terminated", async function() {
		await controlConnector.connect();
		controlConnector._socket.terminate();
		await events.once(controlConnector, "resume");
	});

	it("should reconnect on connection heartbeat timeout", async function() {
		await controlConnector.connect();
		controlConnector.stopHeartbeat();
		await events.once(controlConnector, "resume");
	});

	it("should invalidate on reconnection with bad session token", async function() {
		await controlConnector.connect();
		controlConnector._sessionToken = "blah";
		controlConnector.stopHeartbeat();
		await events.once(controlConnector, "invalidate");
		await events.once(controlConnector, "connect");
	});

	it("should start a new session when the send buffer limit is exceeded", async function() {
		await getControl().send(new lib.ControllerConfigSetFieldRequest("controller.max_send_buffer_size", "0.01"));
		try {
			await controlConnector.connect();
		} finally {
			await getControl().send(new lib.ControllerConfigSetFieldRequest("controller.max_send_buffer_size", "64"));
		}

		// Never acknowledge received messages
		controlConnector._doHeartbeat = function() {
			this._sendInternal(new lib.MessageHeartbeat());
		};
		let invalidated = false;
		controlConnector.once("invalidate", () => { invalidated = true; });
		const connected = events.once(controlConnector, "invalidate").then(
			() => events.once(controlConnector, "connect")
		);
		for (let i = 0; i < 100; i++) {
			if (invalidated) {
				break;
			}
			await control.send(new lib.ControllerConfigGetRequest()).catch(() => {});
		}
		assert(invalidated, "session was not invalidated");
		await connected;
	});

	it("should properly close connector if close is called during reconnect wait", async function() {
		await controlConnector.connect();
		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 = controlToken;
		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 = linkHost.hostId + 1;
		await assert.rejects(
			hostConnector.connect(),
			new Error("Authentication failed: missmatched host id")
		);
	});

	it("should refuse connection for a host that is not registered", async function() {
		hostConnector.token = tokenUnregistered;
		hostConnector.hostId = 0;
		await assert.rejects(
			hostConnector.connect(),
			new Error(
				"Authentication failed: host is not registered, start the host again to request to join the cluster"
			)
		);
	});

	it("should close old connections from the same address", async function() {
		slowTest(this);
		// Without a link the disconnect prepare goes unanswered and the controller waits 10 s for it
		const hostLinkFirst = new lib.Link(new TestHostConnector(url, 0.2));
		const hostConnectorFirst = hostLinkFirst.connector;
		hostConnectorFirst.token = hostConnector.token;
		hostConnectorFirst.hostId = hostConnector.hostId;
		const onceClose = events.once(hostConnectorFirst, "close");
		onceClose.catch(() => {}); // Prevent unhandled promise rejection.
		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: hostId }, Buffer.from("TestSecretDoNotUse", "base64"));
		hostConnector.hostId = hostId; // Same as already running host
		await assert.rejects(
			hostConnector.connect(),
			new Error("Authentication failed: missmatched host id")
		);
	});
});