factorio-clusterio/test/controller/PlayerProxyServer.test.js

import assert from "node:assert/strict";
import dgram from "node:dgram";
import events from "node:events";
import * as lib from "@clusterio/lib";

import { Controller, HostRecord, InstanceRecord, PlayerProxyServer, proxyProtocol } from "@clusterio/controller";

const { MessageType } = proxyProtocol;
const clientVersion = { major: 2, minor: 0, sub: 72, build: 84292 };

/**
 * Encode a ConnectionRequestReplyConfirm the way a Factorio client does
 *
 * @param {object} options - fields of the message.
 * @returns {Buffer[]} the message split into the given number of fragments.
 */
function encodeConfirm({ clientRequestId, serverRequestId, username, messageId = 1, fragments = 1, padding = 40 }) {
	const payload = new proxyProtocol.BufferWriter();
	payload.writeUInt32LE(clientRequestId);
	payload.writeUInt32LE(serverRequestId);
	payload.writeUInt32LE(1234); // instanceID
	const name = Buffer.from(username);
	payload.writeUInt8(name.length);
	for (const byte of name) {
		payload.writeUInt8(byte);
	}
	for (let i = 0; i < padding; i++) {
		payload.writeUInt8(i); // stands in for the password, keys, checksums and mods
	}
	const body = payload.toBuffer();

	if (fragments === 1) {
		return [Buffer.concat([Buffer.from([MessageType.ConnectionRequestReplyConfirm, messageId, 0]), body])];
	}
	const size = Math.ceil(body.length / fragments);
	const result = [];
	for (let i = 0; i < fragments; i++) {
		const last = i === fragments - 1;
		const header = Buffer.from([
			MessageType.ConnectionRequestReplyConfirm | 0x40 | (last ? 0x80 : 0),
			messageId, 0, i,
		]);
		result.push(Buffer.concat([header, body.subarray(i * size, (i + 1) * size)]));
	}
	return result;
}

/** Minimal stand in for a Factorio server's side of the handshake */
class FakeFactorioServer {
	constructor(version = clientVersion) {
		this.version = version;
		this.serverRequestId = 0xabcd1234;
		this.received = [];
		this.socket = dgram.createSocket("udp4");
		this.socket.on("message", (message, rinfo) => this.handle(message, rinfo));
	}

	async start() {
		this.socket.bind(0, "127.0.0.1");
		await events.once(this.socket, "listening");
		this.port = this.socket.address().port;
	}

	handle(message, rinfo) {
		this.received.push(message);
		const type = proxyProtocol.messageType(message);
		if (type === MessageType.ConnectionRequest) {
			const request = proxyProtocol.parseConnectionRequest(message);
			this.socket.send(proxyProtocol.encodeConnectionRequestReply({
				version: this.version,
				clientRequestId: request.clientRequestId,
				serverRequestId: this.serverRequestId,
				maxPacketSize: 508,
			}), rinfo.port, rinfo.address);
		} else if (type === MessageType.ConnectionRequestReplyConfirm) {
			const header = proxyProtocol.parseHeader(message);
			if (!header.fragmented || header.lastFragment) {
				const accept = Buffer.from([MessageType.ConnectionAcceptOrDeny, this.port & 0xff]);
				this.socket.send(accept, rinfo.port, rinfo.address);
			}
		} else if (type === 6) { // ClientToServerHeartbeat
			this.socket.send(Buffer.from([7, ...message.subarray(1)]), rinfo.port, rinfo.address);
		}
	}

	confirms() {
		return this.received.filter(m => proxyProtocol.messageType(m) === MessageType.ConnectionRequestReplyConfirm);
	}

	close() {
		this.socket.close();
	}
}

class FakeClient {
	constructor(proxyPort) {
		this.proxyPort = proxyPort;
		this.socket = dgram.createSocket("udp4");
		this.queue = [];
		this.waiters = [];
		this.socket.on("message", message => {
			const waiter = this.waiters.shift();
			if (waiter) {
				waiter(message);
			} else {
				this.queue.push(message);
			}
		});
	}

	async start() {
		this.socket.bind(0, "127.0.0.1");
		await events.once(this.socket, "listening");
	}

	send(message) {
		this.socket.send(message, this.proxyPort, "127.0.0.1");
	}

	receive(timeoutMs = 2000) {
		if (this.queue.length) {
			return Promise.resolve(this.queue.shift());
		}
		return new Promise((resolve, reject) => {
			const timeout = setTimeout(() => {
				this.waiters.splice(this.waiters.indexOf(waiter), 1);
				reject(new Error("Timed out waiting for packet"));
			}, timeoutMs);
			const waiter = message => {
				clearTimeout(timeout);
				resolve(message);
			};
			this.waiters.push(waiter);
		});
	}

	/** Perform the handshake and return the first packet forwarded from the server */
	async connect(username, { clientRequestId = 42, fragments = 1 } = {}) {
		this.send(proxyProtocol.encodeConnectionRequest(clientVersion, clientRequestId));
		const reply = proxyProtocol.parseConnectionRequestReply(await this.receive());
		for (const fragment of encodeConfirm({
			clientRequestId, serverRequestId: reply.serverRequestId, username, messageId: clientRequestId, fragments,
		})) {
			this.send(fragment);
		}
		return { reply, accept: await this.receive() };
	}

	close() {
		this.socket.close();
	}
}

async function freePort() {
	const socket = dgram.createSocket("udp4");
	socket.bind(0, "127.0.0.1");
	await events.once(socket, "listening");
	const { port } = socket.address();
	socket.close();
	return port;
}

describe("controller/proxy_protocol", function () {
	it("should round trip a connection request", function () {
		const buffer = proxyProtocol.encodeConnectionRequest({ major: 2, minor: 1, sub: 300, build: 90000 }, 7, 5);
		const request = proxyProtocol.parseConnectionRequest(buffer);
		assert.deepEqual(request.version, { major: 2, minor: 1, sub: 300, build: 90000 });
		assert.equal(request.clientRequestId, 7);
		assert.equal(request.header.messageId, 5);
	});

	it("should round trip a connection request reply", function () {
		const reply = { version: clientVersion, clientRequestId: 1, serverRequestId: 0xffffffff, maxPacketSize: 508 };
		const encoded = proxyProtocol.encodeConnectionRequestReply(reply);
		assert.deepEqual(proxyProtocol.parseConnectionRequestReply(encoded), reply);
	});

	it("should parse confirms in the header", function () {
		const buffer = Buffer.from([MessageType.ConnectionRequestReplyConfirm, 3, 0x80, 2, 1, 0, 0, 0, 2, 0, 0, 0]);
		const header = proxyProtocol.parseHeader(buffer);
		assert.equal(header.messageId, 3);
		assert.deepEqual(header.confirms, [1, 2]);
		assert.equal(header.payloadOffset, buffer.length);
	});

	it("should parse the first fragment of a fragmented confirm", function () {
		const [first, second] = encodeConfirm({
			clientRequestId: 1, serverRequestId: 2, username: "Player", fragments: 2,
		});
		const confirm = proxyProtocol.parseConnectionRequestReplyConfirm(first);
		assert.equal(confirm.username, "Player");
		assert.equal(first.readUInt32LE(confirm.serverRequestIdOffset), 2);
		assert.throws(() => proxyProtocol.parseConnectionRequestReplyConfirm(second), /first fragment/);
	});

	it("should reject truncated messages", function () {
		assert.throws(() => proxyProtocol.parseConnectionRequest(Buffer.from([MessageType.ConnectionRequest, 0])));
	});

	it("should build a ping reply", function () {
		const reply = proxyProtocol.pingReplyFor(Buffer.from([0x20, 5, 1]));
		assert.deepEqual(reply, Buffer.from([MessageType.PingReply, 5, 1]));
		assert.equal(proxyProtocol.pingReplyFor(Buffer.from([0])), undefined);
	});
});

describe("controller/PlayerProxyServer", function () {
	let controller;
	let proxy;
	let servers;
	let clients;
	let proxyPort;

	function addInstance(id, server, status = "running", fields = {}) {
		const config = new lib.InstanceConfig("controller", {
			"instance.id": id,
			"instance.name": `Instance ${id}`,
			"instance.assigned_host": 1,
			...fields,
		});
		controller.instances.records.set(new InstanceRecord(config, status, server?.port));
	}

	async function newClient(port = proxyPort) {
		const client = new FakeClient(port);
		await client.start();
		clients.push(client);
		return client;
	}

	beforeEach(async function () {
		const config = new lib.ControllerConfig("controller", { "controller.name": "Test" });
		controller = new Controller(lib.logger, [], config);
		controller.hosts.set(new HostRecord(1, "Host", "2.0.0", new Map(), true, "::ffff:127.0.0.1"));
		servers = [new FakeFactorioServer(), new FakeFactorioServer()];
		await Promise.all(servers.map(server => server.start()));
		addInstance(1, servers[0]);
		addInstance(2, servers[1]);
		clients = [];
		proxyPort = await freePort();
		proxy = new PlayerProxyServer(controller, lib.logger, proxyPort);
		await proxy.start();
	});

	afterEach(async function () {
		await proxy.stop();
		servers.forEach(server => server.close());
		clients.forEach(client => client.close());
	});

	it("should complete the handshake through the proxy", async function () {
		const client = await newClient();
		const { reply, accept } = await client.connect("Player", { fragments: 3 });

		assert.deepEqual(reply.version, clientVersion);
		assert.equal(reply.maxPacketSize, 508);
		assert.equal(accept[0], MessageType.ConnectionAcceptOrDeny);

		const confirms = servers[0].confirms();
		assert.equal(confirms.length, 3);
		const confirm = proxyProtocol.parseConnectionRequestReplyConfirm(confirms[0]);
		assert.equal(confirm.serverRequestId, servers[0].serverRequestId);
		assert.equal(confirm.username, "Player");

		client.send(Buffer.from([6, 1, 2, 3]));
		assert.deepEqual(await client.receive(), Buffer.from([7, 1, 2, 3]));
		assert.equal(proxy.serverVersions.get(1).build, clientVersion.build);
	});

	it("should route by the playerProxyRoute hook", async function () {
		controller.hooks.playerProxyRoute.attach("test", request => {
			assert.deepEqual(request.version, clientVersion);
			return request.username === "Special" ? 2 : undefined;
		});
		const client = await newClient();
		await client.connect("Special");
		assert.equal(servers[1].confirms().length, 1);
		assert.equal(servers[0].confirms().length, 0);
	});

	it("should reroute a new connection from the same socket", async function () {
		let target = 1;
		controller.hooks.playerProxyRoute.attach("test", () => target);
		const client = await newClient();
		await client.connect("Player", { clientRequestId: 1 });
		assert.equal(servers[0].confirms().length, 1);

		target = 2;
		await client.connect("Player", { clientRequestId: 2 });
		assert.equal(servers[1].confirms().length, 1);
	});

	it("should answer pings for clients that are not connected", async function () {
		const client = await newClient();
		client.send(Buffer.from([MessageType.Ping, 9, 0]));
		assert.deepEqual(await client.receive(), Buffer.from([MessageType.PingReply, 9, 0]));
	});

	it("should route dedicated instance ports to their instance", async function () {
		const port = await freePort();
		controller.instances.getMutable(2).config.set("instance.player_proxy_port", port);
		controller.instances.records.set(controller.instances.getMutable(2));
		await proxy.syncListeners();
		assert.equal(proxy.listeners.get(port), 2);

		const client = await newClient(port);
		await client.connect("Player");
		assert.equal(servers[1].confirms().length, 1);

		controller.instances.getMutable(2).config.set("instance.player_proxy_port", null);
		await proxy.syncListeners();
		assert.equal(proxy.listeners.has(port), false);
	});

	it("should start a stopped instance", async function () {
		controller.instances.getMutable(2).status = "stopped";
		controller.instances.getMutable(2).gamePort = undefined;
		controller.hooks.playerProxyRoute.attach("test", () => 2);
		const started = [];
		controller.sendTo = async (dst, message) => {
			assert(message instanceof lib.InstanceStartRequest);
			started.push(dst.instanceId);
			const instance = controller.instances.getMutable(dst.instanceId);
			instance.status = "running";
			instance.gamePort = servers[1].port;
		};
		const client = await newClient();
		await client.connect("Player");
		assert.deepEqual(started, [2]);
	});

	it("should probe the version of a server", async function () {
		servers[0].version = { major: 2, minor: 1, sub: 5, build: 12345 };
		assert.deepEqual(await PlayerProxyServer.probeVersion("127.0.0.1", servers[0].port), servers[0].version);
		assert.deepEqual(await proxy.instanceVersion(1), servers[0].version);
	});
});

describe("controller/Controller.resolvePlayerRoute()", function () {
	let controller;

	beforeEach(function () {
		const config = new lib.ControllerConfig("controller", { "controller.name": "Test" });
		controller = new Controller(lib.logger, [], config);
		for (const [id, status] of [[1, "running"], [2, "stopped"], [3, "stopped"], [4, "stopped"]]) {
			controller.instances.records.set(new InstanceRecord(new lib.InstanceConfig("controller", {
				"instance.id": id, "instance.assigned_host": 1,
			}), status));
		}
		controller.instances.records.set(new InstanceRecord(new lib.InstanceConfig("controller", {
			"instance.id": 5,
		}), "unassigned"));
	});

	const request = (fields = {}) => ({
		username: "player", version: clientVersion, clientAddress: "127.0.0.1", clientPort: 1, listenPort: 1, ...fields,
	});

	it("should prefer hooks, then dedicated port, then last instance, then default", async function () {
		const user = controller.users.getOrCreateUser("Player");
		user.instanceStats.set(3, Object.assign(new lib.PlayerStats(), { lastJoinAt: new Date(2000) }));
		user.instanceStats.set(4, Object.assign(new lib.PlayerStats(), { lastJoinAt: new Date(1000) }));
		controller.config.set("controller.player_proxy_default_instance_id", 4);

		assert.equal(await controller.resolvePlayerRoute(request()), 3);
		assert.equal(await controller.resolvePlayerRoute(request({ username: "other" })), 4);
		assert.equal(await controller.resolvePlayerRoute(request({ portInstanceId: 2 })), 2);
		controller.hooks.playerProxyRoute.attach("test", () => 4);
		assert.equal(await controller.resolvePlayerRoute(request({ portInstanceId: 2 })), 4);
	});

	it("should skip instances that can't be routed to", async function () {
		controller.hooks.playerProxyRoute.attach("test", () => 5);
		assert.equal(await controller.resolvePlayerRoute(request({ portInstanceId: 99 })), 1);
	});
});