diff --git a/docs/configuration.md b/docs/configuration.md index 1818a357..d0d8e607 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -214,6 +214,25 @@ Must be greater than 0. Defaults to 15. +### controller.player_proxy_port + +UDP port the controller accepts connections from Factorio clients on. +Players connecting to it are forwarded to an instance, which is started if it's stopped, see [Player proxy](player-proxy.md). +Disabled when null. +Must be greater than 0. +Requires a restart of the controller to take effect. + +Defaults to null. + + +### controller.player_proxy_default_instance_id + +ID of the instance players connecting through the player proxy are sent to when no plugin picks an instance for them and they have not been on any instance before. +If null, a running instance is picked. + +Defaults to null. + + ### controller.factorio_username Username of the Factorio.com account to authenticate with the Factorio API with. @@ -490,6 +509,15 @@ Does not affect start or stop of the instance while the host is running. Defaults to false. +### instance.player_proxy_port + +Extra UDP port the player proxy on the controller listens on that sends players straight to this instance, starting it if it's stopped. +Only used when controller.player_proxy_port is set, see [Player proxy](player-proxy.md). +Must be greater than 0, and should not be in the host.factorio_port_range of a host on the same machine as the controller. + +Defaults to null. + + ### instance.exclude_from_start_all If enabled this instance will be excluded from the "Start all" button operation in the web interface. diff --git a/packages/controller/index.ts b/packages/controller/index.ts index 6a8937b6..724a91e1 100644 --- a/packages/controller/index.ts +++ b/packages/controller/index.ts @@ -11,6 +11,10 @@ export { default as UserManager } from "./src/UserManager.js"; export { default as UserRecord } from "./src/UserRecord.js"; export { default as User } from "./src/User.js"; export { default as WsServer } from "./src/WsServer.js"; +export { + default as PlayerProxyServer, type PlayerProxyRouteRequest, type PlayerProxyRoute, +} from "./src/PlayerProxyServer.js"; +export * as proxyProtocol from "./src/proxy_protocol.js"; if (import.meta.main) { bootstrap(); diff --git a/packages/controller/src/BaseControllerPlugin.ts b/packages/controller/src/BaseControllerPlugin.ts index 5b5d0874..6a1868a9 100644 --- a/packages/controller/src/BaseControllerPlugin.ts +++ b/packages/controller/src/BaseControllerPlugin.ts @@ -2,6 +2,7 @@ import type Controller from "./Controller.js"; import type InstanceRecord from "./InstanceRecord.js"; import type ControlConnection from "./ControlConnection.js"; import type HostConnection from "./HostConnection.js"; +import type { PlayerProxyRouteRequest } from "./PlayerProxyServer.js"; import * as lib from "@clusterio/lib"; export type ControllerPluginContext = lib.PluginLoadContext<{ @@ -28,6 +29,7 @@ export class ControllerHooks extends lib.AsyncHookCollection { this.modsUpdated = this.newHook(); this.rolesUpdated = this.newHook(); this.playerEvent = this.newHook(); + this.playerProxyRoute = this.newHook(); } /** @@ -266,6 +268,24 @@ export class ControllerHooks extends lib.AsyncHookCollection { * @param event - Information about the event. */ readonly playerEvent: lib.AsyncHook<[instance: InstanceRecord, event: lib.PlayerEvent]>; + + /** + * Called to pick the instance a player connecting through the player proxy goes to + * + * Invoked when a Factorio client has sent its name to the player proxy + * enabled with controller.player_proxy_port. Handlers return the ID of + * the instance to send the player to, or undefined to leave the choice + * to other plugins and the controller. If the instance is stopped the + * proxy starts it, which can take longer than the client is willing to + * wait. + * + * Factorio clients connect twice when joining a server they don't have a + * key for, so a handler must give the same answer for both attempts. + * + * @param request - Name, version and connection details of the player. + * @returns ID of instance to route the player to. + */ + readonly playerProxyRoute: lib.AsyncHook<[request: PlayerProxyRouteRequest], number | undefined>; } // eslint-disable-next-line @typescript-eslint/no-unsafe-declaration-merging diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index 8e455d7b..f51574dd 100644 --- a/packages/controller/src/Controller.ts +++ b/packages/controller/src/Controller.ts @@ -32,8 +32,7 @@ import HostRecord from "./HostRecord.js"; import { BaseControllerPlugin, ControllerHooks } from "./BaseControllerPlugin.js"; import ControllerRouter from "./ControllerRouter.js"; import InstanceManager from "./InstanceManager.js"; -import PlayerRoutingStore from "./PlayerRoutingStore.js"; -import PlayerProxyServer from "./PlayerProxyServer.js"; +import PlayerProxyServer, { type PlayerProxyRouteRequest } from "./PlayerProxyServer.js"; const endpointDurationSummary = new Summary( "clusterio_controller_http_endpoint_duration_seconds", @@ -86,7 +85,6 @@ export default class Controller { trustedProxies: BlockList; debugEvents = new events.EventEmitter(); private _events = new events.EventEmitter(); - playerRoutingStore: PlayerRoutingStore; playerProxyServer: PlayerProxyServer | null = null; /** Event subscription controller */ @@ -290,10 +288,6 @@ export default class Controller { this.clusterLogger = clusterLogger; this.pluginInfos = pluginInfos; this.config = config; - this.playerRoutingStore = new PlayerRoutingStore( - path.join(this.config.get("controller.database_directory"), "player-routing.json"), - ); - this.app = express(); this.app.locals.controller = this; this.app.locals.streams = new Map(); @@ -444,7 +438,6 @@ export default class Controller { // Load plugins await this.loadPlugins(); - await this.playerRoutingStore.load(); const playerProxyPort = this.config.get("controller.player_proxy_port"); if (playerProxyPort !== null) { this.playerProxyServer = new PlayerProxyServer( @@ -816,7 +809,6 @@ export default class Controller { this.users.records.save(), this.hooks.save.invoke(), this.notes.save(), - this.playerRoutingStore.save(), ]); } @@ -1051,87 +1043,65 @@ export default class Controller { return hosts.length ? new lib.HostUpdatesEvent(hosts) : null; } - resolvePlayerRoute(username: string): { hostId: number; instanceId: number; publicAddress: string; gamePort: number } | null { - const key = username.toLowerCase(); - let route = this.playerRoutingStore.get(key); - let instance: Readonly | undefined; - - if (!route) { - const fallbackId = this.config.get("controller.player_proxy_default_instance_id"); - if (fallbackId !== null) { - instance = this.instances.get(fallbackId); - if (instance) { - route = { - name: username, - instanceId: instance.id, - hostId: instance.config.get("instance.assigned_host") ?? -1, - gamePort: instance.gamePort ?? 0, - timestampMs: Date.now(), - }; - } - } else { - instance = this._chooseFallbackInstance(); - if (instance) { - route = { - name: username, - instanceId: instance.id, - hostId: instance.config.get("instance.assigned_host") ?? -1, - gamePort: instance.gamePort ?? 0, - timestampMs: Date.now(), - }; - } - } - } - - if (!route) { - return null; - } - - instance = instance ?? this.instances.get(route.instanceId); - if (!instance || instance.isDeleted) { - this.playerRoutingStore.delete(key); - return null; - } - - const hostId = instance.config.get("instance.assigned_host"); - if (hostId === null || route.hostId !== hostId) { - this.playerRoutingStore.delete(key); - return null; - } - - const host = this.hosts.get(hostId); - if (!host || !host.connected) { - return null; - } - - const gamePort = instance.gamePort ?? route.gamePort; - const publicAddress = host.publicAddress || host.remoteAddress; - if (!gamePort || !publicAddress) { - return null; - } - - return { - hostId, - instanceId: instance.id, - publicAddress, - gamePort, + /** + * Pick the instance a player connecting through the player proxy goes to + * + * Plugins get the first say through the playerProxyRoute hook, then the + * instance the proxy port is dedicated to, the instance the player was + * last on, controller.player_proxy_default_instance_id, and finally any + * running instance. + * + * @param request - Details about the connecting player. + * @returns ID of the instance to route to, or undefined if there is none. + */ + async resolvePlayerRoute(request: PlayerProxyRouteRequest): Promise { + const routable = (id: number | undefined) => { + const instance = id === undefined ? undefined : this.instances.get(id); + return Boolean(instance && !instance.isDeleted && instance.config.get("instance.assigned_host") !== null); }; - } - private _chooseFallbackInstance() { - for (const instance of this.instances.values()) { - if (instance.status === "running" && !instance.isDeleted) { - return instance; + for (const result of await this.hooks.playerProxyRoute.collect(request)) { + if (routable(result)) { + return result; } } + + if (routable(request.portInstanceId)) { + return request.portInstanceId; + } + + const lastInstanceId = this._lastInstanceOfPlayer(request.username); + if (routable(lastInstanceId)) { + return lastInstanceId; + } + + const defaultId = this.config.get("controller.player_proxy_default_instance_id") ?? undefined; + if (routable(defaultId)) { + return defaultId; + } + for (const instance of this.instances.values()) { - if (!instance.isDeleted && instance.config.get("instance.assigned_host") !== null) { - return instance; + if (instance.status === "running") { + return instance.id; } } return undefined; } + private _lastInstanceOfPlayer(name: string) { + const user = this.users.getByName(name); + let lastId: number | undefined; + let lastJoinMs = -Infinity; + for (const [instanceId, stats] of user?.instanceStats ?? []) { + const joinMs = stats.lastJoinAt?.getTime() ?? -Infinity; + if (joinMs > lastJoinMs) { + lastJoinMs = joinMs; + lastId = instanceId; + } + } + return lastId; + } + /** * Removes all saves currently stored for the given instance, if any * @param instanceId - Id of Instance to clear saves for. diff --git a/packages/controller/src/HostConnection.ts b/packages/controller/src/HostConnection.ts index 64495e00..56d5d489 100644 --- a/packages/controller/src/HostConnection.ts +++ b/packages/controller/src/HostConnection.ts @@ -72,7 +72,6 @@ export default class HostConnection extends BaseConnection { this.info.connected = false; this._controller.hosts.set(this.info); - this._controller.playerRoutingStore.dropHost(this.id); }); this.handle(lib.GetFactorioCredentialsRequest, this.handleGetFactorioCredentialsRequest.bind(this)); @@ -82,7 +81,6 @@ export default class HostConnection extends BaseConnection { this.handle(lib.InstanceSaveDetailsUpdatesEvent, this.handleInstanceSaveDetailsUpdatesEvent.bind(this)); this.handle(lib.LogMessageEvent, this.handleLogMessageEvent.bind(this)); this.handle(lib.InstancePlayerUpdateEvent, this.handleInstancePlayerUpdateEvent.bind(this)); - this.handle(lib.InstancePlayerRouteUpdateEvent, this.handleInstancePlayerRouteUpdateEvent.bind(this)); this.snoopEvent(lib.InstanceAdminlistUpdateEvent, this.snoopInstanceAdminlistUpdateEvent.bind(this)); this.snoopEvent(lib.InstanceBanlistUpdateEvent, this.snoopInstanceBanlistUpdateEvent.bind(this)); this.snoopEvent(lib.InstanceWhitelistUpdateEvent, this.snoopInstanceWhitelistUpdateEvent.bind(this)); @@ -413,30 +411,6 @@ export default class HostConnection extends BaseConnection { }); } - async handleInstancePlayerRouteUpdateEvent(event: lib.InstancePlayerRouteUpdateEvent, src: lib.Address) { - const instanceId = src.id; - if (event.instanceId !== instanceId) { - logger.warn( - `Received route update for instance ${event.instanceId} from ${instanceId}, ignoring mismatched id` - ); - return; - } - if (event.hostId !== this.id) { - logger.warn( - `Received route update for host ${event.hostId} from ${this.id}, ignoring mismatched host` - ); - return; - } - - this._controller.playerRoutingStore.update( - event.name, - instanceId, - this.id, - event.gamePort, - event.timestampMs, - ); - } - async snoopInstanceAdminlistUpdateEvent(event: lib.InstanceAdminlistUpdateEvent, src: lib.Address) { if (src.type !== lib.Address.instance) { return; diff --git a/packages/controller/src/PlayerProxyServer.ts b/packages/controller/src/PlayerProxyServer.ts index 284625f1..b53b77aa 100644 --- a/packages/controller/src/PlayerProxyServer.ts +++ b/packages/controller/src/PlayerProxyServer.ts @@ -1,153 +1,105 @@ -import dgram from "dgram"; +import dgram from "node:dgram"; +import { isIPv6 } from "node:net"; import * as lib from "@clusterio/lib"; import type Controller from "./Controller.js"; - -const TYPE_MASK = 0x1f; -const FLAG_MASK = 0xe0; - -const TYPE_CONNECTION_REQUEST = 2; -const TYPE_CONNECTION_REQUEST_REPLY = 3; -const TYPE_CONNECTION_REQUEST_REPLY_CONFIRM = 4; +import type InstanceRecord from "./InstanceRecord.js"; +import * as protocol from "./proxy_protocol.js"; +import { MessageType } from "./proxy_protocol.js"; const SESSION_IDLE_TIMEOUT_MS = 30_000; -const MAX_PENDING_PACKETS = 64; -const MAX_PACKET_SIZE = 1400; +const INSTANCE_START_TIMEOUT_MS = 120_000; +const UPSTREAM_RETRY_INTERVAL_MS = 500; +const UPSTREAM_RETRY_COUNT = 20; +const PROBE_TIMEOUT_MS = 2_000; +const MAX_PENDING_PACKETS = 256; -type ClientKey = string; +/** Details about a player connecting through the proxy that is used to decide where to send them */ +export interface PlayerProxyRouteRequest { + /** Name the player connects with */ + username: string; + /** Factorio version and build of the connecting client */ + version: protocol.ApplicationVersion; + /** Address and port the client connects from */ + clientAddress: string; + clientPort: number; + /** Proxy port the client connected to */ + listenPort: number; + /** Instance the proxy port is dedicated to, if any */ + portInstanceId?: number; +} -type RouteTarget = { - hostId: number; +export interface PlayerProxyRoute { instanceId: number; - publicAddress: string; - gamePort: number; + address: string; + port: number; +} + +type Listener = { + port: number; + socket: dgram.Socket; + instanceId?: number; }; type ClientSession = { + key: string; + listener: Listener; clientAddress: string; clientPort: number; lastSeenMs: number; - pending: Buffer[]; - upstream: dgram.Socket | null; - route: RouteTarget | null; - requestBuffer?: Buffer; - requestFlags?: number; - requestMessageIdAndFlags?: number; - requestVersion?: ApplicationVersion; - clientRequestId?: number; - proxyServerId?: number; - pendingConfirm?: Buffer; - confirmServerIdOffset?: number; - serverConnectionId?: number; + /** Raw ConnectionRequest received from the client */ + request?: protocol.ConnectionRequest & { buffer: Buffer }; + /** Request id the proxy answered the client with */ + proxyServerRequestId: number; username?: string; - activating?: Promise; + /** Fragments of the client's ConnectionRequestReplyConfirm by fragment number */ + confirmFragments: Map; + confirmMessageId?: number; + confirmServerRequestIdOffset?: number; + state: "handshake" | "routing" | "connecting" | "connected" | "failed"; + route?: PlayerProxyRoute; + upstream?: dgram.Socket; + upstreamRetry?: ReturnType; + serverRequestId?: number; + /** Packets received from the client before the upstream connection was ready */ + pending: Buffer[]; }; -type ApplicationVersion = { - major: number; - minor: number; - sub: number; - build: number; -}; - -class BufferReader { - private _offset = 0; - constructor(private _buffer: Buffer) {} - - get offset() { - return this._offset; - } - - readUInt8() { - const value = this._buffer.readUInt8(this._offset); - this._offset += 1; - return value; - } - - readUInt16LE() { - const value = this._buffer.readUInt16LE(this._offset); - this._offset += 2; - return value; - } - - readUInt32LE() { - const value = this._buffer.readUInt32LE(this._offset); - this._offset += 4; - return value; - } - - readSpaceOptimized() { - const value = this.readUInt8(); - if (value === 0xff) { - return this.readUInt16LE(); - } - return value; - } - - readContainerSize() { - const value = this.readUInt8(); - if (value === 0xff) { - return this.readUInt32LE(); - } - return value; - } - - readBytes(length: number) { - const start = this._offset; - const end = start + length; - if (end > this._buffer.length) { - throw new Error("Buffer too short"); - } - this._offset = end; - return this._buffer.slice(start, end); - } +function randomUInt32() { + return Math.floor(Math.random() * 2 ** 32) >>> 0; } -class BufferWriter { - private _parts: Buffer[] = []; - - writeUInt8(value: number) { - const buf = Buffer.allocUnsafe(1); - buf.writeUInt8(value, 0); - this._parts.push(buf); - } - - writeUInt16LE(value: number) { - const buf = Buffer.allocUnsafe(2); - buf.writeUInt16LE(value, 0); - this._parts.push(buf); - } - - writeUInt32LE(value: number) { - const buf = Buffer.allocUnsafe(4); - buf.writeUInt32LE(value >>> 0, 0); - this._parts.push(buf); - } - - writeSpaceOptimized(value: number) { - if (value >= 0xff) { - this.writeUInt8(0xff); - this.writeUInt16LE(value); - } else { - this.writeUInt8(value); - } - } - - concat() { - return Buffer.concat(this._parts); - } -} - -function clientKey(address: string, port: number): ClientKey { - return `${address}:${port}`; +/** + * Strip the IPv4-mapped prefix from addresses so they can be used with udp4 sockets + * @param address - Address to normalise. + * @returns The address with any ::ffff: prefix removed. + */ +function normaliseAddress(address: string) { + return address.startsWith("::ffff:") ? address.slice(7) : address; } +/** + * UDP proxy for Factorio players + * + * Accepts connections from Factorio clients, completes the first step of the + * connection handshake itself to learn the name and version of the player, + * and then forwards the connection to the instance picked by the controller. + * Stopped instances are started on demand. + */ export default class PlayerProxyServer { - private _socket: dgram.Socket | null = null; - private _sessions = new Map(); + private _listeners = new Map(); + private _sessions = new Map(); private _cleanupTimer: ReturnType | undefined; private _closed = false; + private _onInstancesUpdated = () => { + this.syncListeners().catch(err => { + this._logger.error(`Unexpected error syncing proxy ports:\n${err.stack}`); + }); + }; + + /** Last version and build seen from each instance's Factorio server */ + readonly serverVersions = new Map(); constructor( private _controller: Controller, @@ -155,28 +107,25 @@ export default class PlayerProxyServer { private _port: number, ) {} + get port() { + return this._port; + } + + /** Ports currently listened on, mapped to the instance they are dedicated to */ + get listeners(): ReadonlyMap { + return new Map([...this._listeners.values()].map(listener => [listener.port, listener.instanceId])); + } + async start() { - if (this._socket || this._closed) { + if (this._closed) { return; } - - const socket = dgram.createSocket("udp4"); - this._socket = socket; - socket.on("message", this._handleClientMessage.bind(this)); - socket.on("error", err => { - this._logger.error(`Player proxy socket error: ${err.message}`); - this.stop().catch(() => {}); - }); - - await new Promise((resolve, reject) => { - socket.once("listening", resolve); - socket.once("error", reject); - socket.bind(this._port, "0.0.0.0"); - }); + await this._listen(this._port); + this._controller.instances.records.on("update", this._onInstancesUpdated); + await this.syncListeners(); this._cleanupTimer = setInterval(() => this._cleanupSessions(), SESSION_IDLE_TIMEOUT_MS); this._cleanupTimer.unref(); - this._logger.info(`Player proxy listening on UDP ${this._port}`); } async stop() { @@ -184,302 +133,449 @@ export default class PlayerProxyServer { return; } this._closed = true; + this._controller.instances.records.off("update", this._onInstancesUpdated); - if (this._cleanupTimer) { - clearInterval(this._cleanupTimer); - this._cleanupTimer = undefined; - } - + clearInterval(this._cleanupTimer); for (const session of this._sessions.values()) { - session.upstream?.close(); + this._closeSession(session); } this._sessions.clear(); - if (this._socket) { - await new Promise(resolve => this._socket!.close(() => resolve())); - this._socket = null; - } + await Promise.all([...this._listeners.values()].map(listener => this._closeListener(listener))); } - private _handleClientMessage(message: Buffer, rinfo: dgram.RemoteInfo) { - if (this._closed || !this._socket) { + /** + * Open and close dedicated instance ports to match instance.player_proxy_port + */ + async syncListeners() { + await this._syncListenersSerial.invoke(); + } + + private _syncListenersSerial = new lib.AsyncSerialMergingCallback(async () => { + if (this._closed) { return; } - const key = clientKey(rinfo.address, rinfo.port); + const wanted = new Map(); + for (const instance of this._controller.instances.values()) { + const port = instance.config.get("instance.player_proxy_port"); + if (port === null || instance.isDeleted) { + continue; + } + if (port === this._port) { + this._logger.warn(`Instance ${instance.id} uses the main player proxy port ${port}, ignoring it`); + continue; + } + if (wanted.has(port)) { + this._logger.warn( + `Instances ${wanted.get(port)} and ${instance.id} both use player proxy port ${port}, ` + + `only routing it to ${wanted.get(port)}` + ); + continue; + } + wanted.set(port, instance.id); + } + + for (const listener of [...this._listeners.values()]) { + if (listener.port === this._port) { + continue; + } + const instanceId = wanted.get(listener.port); + if (instanceId === undefined) { + await this._closeListener(listener); + } else { + listener.instanceId = instanceId; + } + } + + for (const [port, instanceId] of wanted) { + if (!this._listeners.has(port)) { + try { + await this._listen(port, instanceId); + } catch (err: any) { + this._logger.error(`Failed to listen on player proxy port ${port}: ${err.message}`); + } + } + } + }); + + private async _listen(port: number, instanceId?: number) { + const socket = dgram.createSocket("udp4"); + const listener: Listener = { port, socket, instanceId }; + await new Promise((resolve, reject) => { + socket.once("error", reject); + socket.bind(port, () => { + socket.off("error", reject); + resolve(); + }); + }); + socket.on("error", err => { + this._logger.error(`Player proxy socket on port ${port} errored: ${err.message}`); + }); + socket.on("message", (message, rinfo) => { + try { + this._handleClientMessage(listener, message, rinfo); + } catch (err: any) { + this._logger.error(`Unexpected error handling player proxy packet:\n${err.stack}`); + } + }); + this._listeners.set(port, listener); + this._logger.info( + instanceId === undefined + ? `Player proxy listening on UDP ${port}` + : `Player proxy listening on UDP ${port} for instance ${instanceId}` + ); + } + + private async _closeListener(listener: Listener) { + this._listeners.delete(listener.port); + for (const session of [...this._sessions.values()]) { + if (session.listener === listener) { + this._closeSession(session); + } + } + await new Promise(resolve => listener.socket.close(() => resolve())); + } + + private _handleClientMessage(listener: Listener, message: Buffer, rinfo: dgram.RemoteInfo) { + if (this._closed) { + return; + } + + const type = protocol.messageType(message); + const key = `${listener.port}|${rinfo.address}|${rinfo.port}`; let session = this._sessions.get(key); + + if (type === MessageType.ConnectionRequest) { + let request: protocol.ConnectionRequest; + try { + request = protocol.parseConnectionRequest(message); + } catch (err: any) { + this._logger.verbose(`Ignoring malformed connection request: ${err.message}`); + return; + } + + // A different request id means the client started a new connection from the same socket, + // which is what Factorio does after LuaPlayer::connect_to_server. + if (session && session.request && session.request.clientRequestId !== request.clientRequestId) { + this._closeSession(session); + session = undefined; + } + if (!session) { + session = this._createSession(key, listener, rinfo); + } + session.lastSeenMs = Date.now(); + this._handleConnectionRequest(session, request, message); + return; + } + if (!session) { - session = this._createSession(rinfo.address, rinfo.port); - this._sessions.set(key, session); + if (type === MessageType.Ping) { + // Pings from the server browser are answered for the proxy itself + this._replyToPing(listener, message, rinfo); + } + return; } session.lastSeenMs = Date.now(); - let type: number; - try { - type = message.readUInt8(0) & TYPE_MASK; - } catch { + if (type === MessageType.ConnectionRequestReplyConfirm) { + this._handleConfirm(session, message); return; } - if (type === TYPE_CONNECTION_REQUEST) { - this._handleConnectionRequest(session, message); - return; - } - - if (type === TYPE_CONNECTION_REQUEST_REPLY_CONFIRM) { - this._handleConnectionRequestReplyConfirm(session, message); - return; - } - - if (!session.route || !session.upstream) { + if (session.state === "connected") { + this._sendUpstream(session, message); + } else if (type === MessageType.Ping) { + this._replyToPing(listener, message, rinfo); + } else { this._queuePending(session, message); - return; } - - this._flushPending(session); - session.upstream.send(message, session.route.gamePort, session.route.publicAddress); } - private _createSession(address: string, port: number): ClientSession { - return { - clientAddress: address, - clientPort: port, + private _replyToPing(listener: Listener, message: Buffer, rinfo: dgram.RemoteInfo) { + const reply = protocol.pingReplyFor(message); + if (reply) { + listener.socket.send(reply, rinfo.port, rinfo.address); + } + } + + private _createSession(key: string, listener: Listener, rinfo: dgram.RemoteInfo): ClientSession { + const session: ClientSession = { + key, + listener, + clientAddress: rinfo.address, + clientPort: rinfo.port, lastSeenMs: Date.now(), + proxyServerRequestId: randomUInt32(), + confirmFragments: new Map(), + state: "handshake", pending: [], - upstream: null, - route: null, }; + this._sessions.set(key, session); + return session; } - private _handleConnectionRequest(session: ClientSession, message: Buffer) { - let parsed: { - flags: number; - messageIdAndFlags: number; - version: ApplicationVersion; - requestId: number; - }; - try { - parsed = this._parseConnectionRequest(message); - } catch (err: any) { - this._logger.warn(`Failed to parse connection request: ${err.message ?? err}`); + private _handleConnectionRequest(session: ClientSession, request: protocol.ConnectionRequest, buffer: Buffer) { + session.request = { ...request, buffer }; + if (session.state !== "handshake" && session.state !== "routing") { return; } - session.requestBuffer = message; - session.requestFlags = parsed.flags; - session.requestMessageIdAndFlags = parsed.messageIdAndFlags; - session.requestVersion = parsed.version; - session.clientRequestId = parsed.requestId; - if (!session.proxyServerId) { - session.proxyServerId = Math.floor(Math.random() * 2 ** 32); - } - - const reply = this._buildConnectionRequestReply(session); - this._socket?.send(reply, session.clientPort, session.clientAddress); + // Reply with the client's own version so it proceeds to send its username. If the instance + // it ends up on runs a different version the connection will fail at that point instead. + this._sendToClient(session, protocol.encodeConnectionRequestReply({ + version: request.version, + clientRequestId: request.clientRequestId, + serverRequestId: session.proxyServerRequestId, + maxPacketSize: protocol.DEFAULT_MAX_PACKET_SIZE, + })); } - private _handleConnectionRequestReplyConfirm(session: ClientSession, message: Buffer) { - let parsed: { username: string; serverIdOffset: number }; + private _handleConfirm(session: ClientSession, message: Buffer) { + let header: protocol.MessageHeader; try { - parsed = this._parseConnectionRequestReplyConfirm(message); + header = protocol.parseHeader(message); } catch (err: any) { - this._logger.warn(`Failed to parse connection confirm: ${err.message ?? err}`); + this._logger.verbose(`Ignoring malformed connection confirm: ${err.message}`); return; } - session.username = parsed.username; - session.pendingConfirm = message; - session.confirmServerIdOffset = parsed.serverIdOffset; - - if (session.route || session.activating) { + const fragmentNumber = header.fragmented ? header.fragmentNumber! : 0; + if (fragmentNumber === 0) { + let confirm: protocol.ConnectionRequestReplyConfirm; + try { + confirm = protocol.parseConnectionRequestReplyConfirm(message); + } catch (err: any) { + this._logger.verbose(`Ignoring malformed connection confirm: ${err.message}`); + return; + } + if ( + confirm.serverRequestId !== session.proxyServerRequestId + && confirm.serverRequestId !== session.serverRequestId + ) { + return; + } + if (session.confirmMessageId !== undefined && session.confirmMessageId !== header.messageId) { + // The client is sending a different confirm message, start over with it + session.confirmFragments.clear(); + } + session.confirmMessageId = header.messageId; + session.confirmServerRequestIdOffset = confirm.serverRequestIdOffset; + session.username = confirm.username; + } else if (session.confirmMessageId !== undefined && header.messageId !== session.confirmMessageId) { return; } - session.activating = this._activateRoute(session).finally(() => { - session.activating = undefined; + if (session.serverRequestId !== undefined) { + // Resends of the confirm are passed on directly once the instance has answered + this._sendUpstream(session, this._patchConfirm(session, message, fragmentNumber)); + return; + } + + if (session.confirmFragments.size < MAX_PENDING_PACKETS) { + session.confirmFragments.set(fragmentNumber, message); + } + + if (session.state === "handshake" && session.username !== undefined && session.request) { + session.state = "routing"; + this._routeSession(session).catch(err => { + this._logger.error(`Unexpected error routing player ${session.username}:\n${err.stack}`); + session.state = "failed"; + }); + } + } + + private _patchConfirm(session: ClientSession, message: Buffer, fragmentNumber: number) { + if (fragmentNumber !== 0 || session.confirmServerRequestIdOffset === undefined) { + return message; + } + const patched = Buffer.from(message); + patched.writeUInt32LE(session.serverRequestId!, session.confirmServerRequestIdOffset); + return patched; + } + + private async _routeSession(session: ClientSession) { + const request: PlayerProxyRouteRequest = { + username: session.username!, + version: session.request!.version, + clientAddress: session.clientAddress, + clientPort: session.clientPort, + listenPort: session.listener.port, + portInstanceId: session.listener.instanceId, + }; + + const instanceId = await this._controller.resolvePlayerRoute(request); + if (this._sessions.get(session.key) !== session) { + return; // Session was replaced while routing + } + if (instanceId === undefined) { + this._logger.warn(`No instance to route player ${request.username} to`); + session.state = "failed"; + return; + } + + const route = await this._resolveInstanceAddress(instanceId); + if (this._sessions.get(session.key) !== session) { + return; + } + if (!route) { + this._logger.warn(`Unable to route player ${request.username} to instance ${instanceId}`); + session.state = "failed"; + return; + } + + this._logger.verbose( + `Routing player ${request.username} (Factorio ${protocol.formatVersion(request.version)}) ` + + `to instance ${instanceId} at ${route.address}:${route.port}` + ); + session.route = route; + session.state = "connecting"; + const upstream = dgram.createSocket(isIPv6(route.address) ? "udp6" : "udp4"); + session.upstream = upstream; + upstream.on("message", data => { + try { + this._handleUpstreamMessage(session, data); + } catch (err: any) { + this._logger.error(`Unexpected error handling upstream packet:\n${err.stack}`); + } + }); + upstream.on("error", err => { + this._logger.warn(`Upstream socket for ${request.username} errored: ${err.message}`); + this._closeSession(session); }); - } - private _buildConnectionRequestReply(session: ClientSession) { - const flags = session.requestFlags ?? 0; - const version = session.requestVersion ?? { major: 0, minor: 0, sub: 0, build: 0 }; - const requestId = session.clientRequestId ?? 0; - const serverId = session.proxyServerId ?? 0; - - const writer = new BufferWriter(); - writer.writeUInt8((flags & FLAG_MASK) | TYPE_CONNECTION_REQUEST_REPLY); - writer.writeSpaceOptimized(version.major); - writer.writeSpaceOptimized(version.minor); - writer.writeSpaceOptimized(version.sub); - writer.writeUInt32LE(version.build); - writer.writeUInt32LE(requestId); - writer.writeUInt32LE(serverId); - writer.writeUInt32LE(MAX_PACKET_SIZE); - return writer.concat(); - } - - private _parseConnectionRequest(buffer: Buffer) { - const reader = new BufferReader(buffer); - const header = reader.readUInt8(); - const type = header & TYPE_MASK; - if (type !== TYPE_CONNECTION_REQUEST) { - throw new Error("Unexpected message type"); - } - - const flags = header & FLAG_MASK; - const messageIdAndFlags = reader.readUInt16LE(); - const version = { - major: reader.readSpaceOptimized(), - minor: reader.readSpaceOptimized(), - sub: reader.readSpaceOptimized(), - build: reader.readUInt32LE(), + let attempts = 0; + const sendRequest = () => { + attempts += 1; + if (attempts > UPSTREAM_RETRY_COUNT) { + this._logger.warn(`Instance ${instanceId} did not answer connection of ${request.username}`); + this._closeSession(session); + return; + } + this._sendUpstream(session, session.request!.buffer); }; - const requestId = reader.readUInt32LE(); - return { flags, messageIdAndFlags, version, requestId }; + sendRequest(); + session.upstreamRetry = setInterval(sendRequest, UPSTREAM_RETRY_INTERVAL_MS); } - private _parseConnectionRequestReplyConfirm(buffer: Buffer) { - const reader = new BufferReader(buffer); - const header = reader.readUInt8(); - const type = header & TYPE_MASK; - if (type !== TYPE_CONNECTION_REQUEST_REPLY_CONFIRM) { - throw new Error("Unexpected message type"); + /** + * Find the address of an instance, starting it if it is stopped. + * + * @param instanceId - ID of the instance to look up. + * @returns address and port of the instance, or undefined if it's not reachable. + */ + private async _resolveInstanceAddress(instanceId: number): Promise { + let instance = await this._ensureInstanceRunning(instanceId); + if (!instance) { + return undefined; } - reader.readUInt16LE(); - reader.readUInt32LE(); // client request id - const serverIdOffset = reader.offset; - reader.readUInt32LE(); - reader.readUInt32LE(); // instance id - const nameLength = reader.readContainerSize(); - const nameBuffer = reader.readBytes(nameLength); - const username = nameBuffer.toString("utf8"); - return { username, serverIdOffset }; + const hostId = instance.config.get("instance.assigned_host"); + const host = hostId !== null ? this._controller.hosts.get(hostId) : undefined; + if (!host || !host.connected || !instance.gamePort) { + return undefined; + } + // The address the host connected to the controller from is preferred as it is how the controller + // reaches the host, the public address may not be reachable from inside the network. + const address = normaliseAddress(host.remoteAddress || host.publicAddress); + if (!address) { + return undefined; + } + return { instanceId, address, port: instance.gamePort }; + } + + private async _ensureInstanceRunning(instanceId: number): Promise | undefined> { + const instance = this._controller.instances.get(instanceId); + if (!instance || instance.isDeleted || instance.config.get("instance.assigned_host") === null) { + return undefined; + } + if (instance.status === "running") { + return instance; + } + + if (instance.status === "stopped") { + this._logger.info(`Starting instance ${instanceId} for a player connecting through the proxy`); + try { + await this._controller.sendTo({ instanceId }, new lib.InstanceStartRequest()); + } catch (err: any) { + this._logger.warn(`Failed to start instance ${instanceId}: ${err.message}`); + return undefined; + } + } + + const deadline = Date.now() + INSTANCE_START_TIMEOUT_MS; + while (Date.now() < deadline && !this._closed) { + const current = this._controller.instances.get(instanceId); + if (!current || current.isDeleted) { + return undefined; + } + if (current.status === "running") { + return current; + } + if (current.status === "stopped") { + return undefined; + } + await lib.wait(500); + } + return undefined; } private _handleUpstreamMessage(session: ClientSession, data: Buffer) { - if (!this._socket || !session.route) { + if (this._sessions.get(session.key) !== session) { return; } - let type: number; - try { - type = data.readUInt8(0) & TYPE_MASK; - } catch { - return; - } - - if (type === TYPE_CONNECTION_REQUEST_REPLY && session.serverConnectionId === undefined) { + if (protocol.messageType(data) === MessageType.ConnectionRequestReply) { + if (session.serverRequestId !== undefined) { + return; + } + let reply: protocol.ConnectionRequestReply; try { - const serverConnectionId = this._parseConnectionRequestReply(data); - session.serverConnectionId = serverConnectionId; - this._sendPendingConfirm(session); + reply = protocol.parseConnectionRequestReply(data); } catch (err: any) { - this._logger.warn(`Failed to parse upstream reply: ${err.message ?? err}`); - } - return; - } - - this._socket.send(data, session.clientPort, session.clientAddress); - } - - private _parseConnectionRequestReply(buffer: Buffer) { - const reader = new BufferReader(buffer); - const header = reader.readUInt8(); - const type = header & TYPE_MASK; - if (type !== TYPE_CONNECTION_REQUEST_REPLY) { - throw new Error("Unexpected message type"); - } - reader.readSpaceOptimized(); - reader.readSpaceOptimized(); - reader.readSpaceOptimized(); - reader.readUInt32LE(); - reader.readUInt32LE(); // client request id - return reader.readUInt32LE(); - } - - private async _activateRoute(session: ClientSession) { - if (!session.username) { - return; - } - - const route = this._controller.resolvePlayerRoute(session.username); - if (!route) { - this._logger.warn(`No route for ${session.username}, dropping client ${session.clientAddress}:${session.clientPort}`); - return; - } - - const instanceReady = await this._ensureInstanceRunning(route.instanceId); - if (!instanceReady) { - this._logger.warn(`Instance ${route.instanceId} not ready for ${session.username}`); - return; - } - - session.route = route; - if (!session.upstream) { - const upstream = dgram.createSocket("udp4"); - session.upstream = upstream; - upstream.on("message", data => this._handleUpstreamMessage(session, data)); - upstream.on("error", err => { this._logger.warn( - `Upstream error for ${session.clientAddress}:${session.clientPort}: ${err.message}` + `Ignoring malformed reply from instance ${session.route!.instanceId}: ${err.message}` ); - upstream.close(); - this._sessions.delete(clientKey(session.clientAddress, session.clientPort)); - }); - } - - if (session.requestBuffer) { - session.upstream.send(session.requestBuffer, route.gamePort, route.publicAddress); - } - } - - private async _ensureInstanceRunning(instanceId: number) { - const instance = this._controller.instances.get(instanceId); - if (!instance || instance.isDeleted) { - return false; - } - if (instance.status === "running") { - return true; - } - if (instance.config.get("instance.assigned_host") === null) { - return false; - } - - try { - await this._controller.sendTo({ instanceId }, new lib.InstanceStartRequest()); - } catch (err: any) { - this._logger.warn(`Failed to start instance ${instanceId}: ${err.message ?? err}`); - return false; - } - - const deadline = Date.now() + SESSION_IDLE_TIMEOUT_MS; - while (Date.now() < deadline) { - const current = this._controller.instances.get(instanceId); - if (!current || current.isDeleted) { - return false; + return; } - if (current.status === "running") { - return true; + if (reply.clientRequestId !== session.request!.clientRequestId) { + return; } - await new Promise(resolve => setTimeout(resolve, 500)); - } - return false; - } - private _sendPendingConfirm(session: ClientSession) { - if (!session.pendingConfirm || session.confirmServerIdOffset === undefined || !session.upstream || !session.route) { + clearInterval(session.upstreamRetry); + session.upstreamRetry = undefined; + session.serverRequestId = reply.serverRequestId; + this.serverVersions.set(session.route!.instanceId, reply.version); + if (!protocol.versionsEqual(reply.version, session.request!.version)) { + this._logger.warn( + `Player ${session.username} runs Factorio ${protocol.formatVersion(session.request!.version)} ` + + `(build ${session.request!.version.build}) but instance ${session.route!.instanceId} runs ` + + `${protocol.formatVersion(reply.version)} (build ${reply.version.build})` + ); + } + + for (const [fragmentNumber, fragment] of [...session.confirmFragments].sort(([a], [b]) => a - b)) { + this._sendUpstream(session, this._patchConfirm(session, fragment, fragmentNumber)); + } + session.confirmFragments.clear(); + session.state = "connected"; + for (const pending of session.pending) { + this._sendUpstream(session, pending); + } + session.pending = []; return; } - if (session.serverConnectionId === undefined) { - return; - } + this._sendToClient(session, data); + } - const payload = Buffer.from(session.pendingConfirm); - payload.writeUInt32LE(session.serverConnectionId, session.confirmServerIdOffset); - session.upstream.send(payload, session.route.gamePort, session.route.publicAddress); - session.pendingConfirm = undefined; - this._flushPending(session); + private _sendToClient(session: ClientSession, data: Buffer) { + session.listener.socket.send(data, session.clientPort, session.clientAddress); + } + + private _sendUpstream(session: ClientSession, data: Buffer) { + if (session.upstream && session.route) { + session.upstream.send(data, session.route.port, session.route.address); + } } private _queuePending(session: ClientSession, message: Buffer) { @@ -489,23 +585,82 @@ export default class PlayerProxyServer { session.pending.push(message); } - private _flushPending(session: ClientSession) { - if (!session.route || !session.upstream || !session.pending.length) { - return; + private _closeSession(session: ClientSession) { + clearInterval(session.upstreamRetry); + session.upstreamRetry = undefined; + if (session.upstream) { + session.upstream.close(); + session.upstream = undefined; } - for (const pending of session.pending) { - session.upstream.send(pending, session.route.gamePort, session.route.publicAddress); + if (this._sessions.get(session.key) === session) { + this._sessions.delete(session.key); } - session.pending = []; } private _cleanupSessions() { const now = Date.now(); - for (const [key, session] of this._sessions) { + for (const session of [...this._sessions.values()]) { if (now - session.lastSeenMs > SESSION_IDLE_TIMEOUT_MS) { - session.upstream?.close(); - this._sessions.delete(key); + this._closeSession(session); } } } + + /** + * Ask a Factorio server which version and build it runs + * + * Sends a connection request and reads the version from the reply, the + * server forgets the request after its connection timeout. + * + * @param address - Address of the Factorio server. + * @param port - Game port of the Factorio server. + * @returns Version of the server, or undefined if it did not answer. + */ + static async probeVersion(address: string, port: number): Promise { + const socket = dgram.createSocket(isIPv6(address) ? "udp6" : "udp4"); + const clientRequestId = randomUInt32(); + try { + return await new Promise((resolve, reject) => { + const timeout = setTimeout(() => resolve(undefined), PROBE_TIMEOUT_MS); + socket.on("error", err => { + clearTimeout(timeout); + reject(err); + }); + socket.on("message", data => { + try { + const reply = protocol.parseConnectionRequestReply(data); + if (reply.clientRequestId === clientRequestId) { + clearTimeout(timeout); + resolve(reply.version); + } + } catch { /* not the reply we're waiting for */ } + }); + // The server replies with its own version regardless of what the client sends. + const version = { major: 0, minor: 0, sub: 0, build: 0 }; + socket.send(protocol.encodeConnectionRequest(version, clientRequestId), port, address); + }); + } finally { + socket.close(); + } + } + + /** + * Find the version and build of an instance + * + * @param instanceId - ID of the instance. + * @returns version of the running instance, or the last version seen if it is stopped. + */ + async instanceVersion(instanceId: number) { + const instance = this._controller.instances.get(instanceId); + if (instance?.status === "running") { + const route = await this._resolveInstanceAddress(instanceId); + if (route) { + const version = await PlayerProxyServer.probeVersion(route.address, route.port).catch(() => undefined); + if (version) { + this.serverVersions.set(instanceId, version); + } + } + } + return this.serverVersions.get(instanceId); + } } diff --git a/packages/controller/src/PlayerRoutingStore.ts b/packages/controller/src/PlayerRoutingStore.ts deleted file mode 100644 index 33fba7bd..00000000 --- a/packages/controller/src/PlayerRoutingStore.ts +++ /dev/null @@ -1,75 +0,0 @@ -import fs from "node:fs/promises"; - -import * as lib from "@clusterio/lib"; - -export type StoredRoute = { - name: string; - instanceId: number; - hostId: number; - gamePort: number; - timestampMs: number; -}; - -export default class PlayerRoutingStore { - private _routes = new Map(); - private _dirty = false; - - constructor(private _filePath: string) {} - - get(name: string): StoredRoute | undefined { - return this._routes.get(name.toLowerCase()); - } - - update(name: string, instanceId: number, hostId: number, gamePort: number, timestampMs: number) { - const key = name.toLowerCase(); - this._routes.set(key, { name, instanceId, hostId, gamePort, timestampMs }); - this._dirty = true; - } - - delete(name: string) { - const key = name.toLowerCase(); - if (this._routes.delete(key)) { - this._dirty = true; - } - } - - dropHost(hostId: number) { - let removed = false; - for (const [key, route] of this._routes) { - if (route.hostId === hostId) { - this._routes.delete(key); - removed = true; - } - } - if (removed) { - this._dirty = true; - } - } - - async load(): Promise { - try { - const routes = JSON.parse(await fs.readFile(this._filePath, { encoding: "utf8" })) as StoredRoute[]; - this._routes.clear(); - for (const route of routes) { - if (!route || !route.name) { - continue; - } - this._routes.set(route.name.toLowerCase(), route); - } - this._dirty = false; - } catch (err: any) { - if (err.code !== "ENOENT") { - throw err; - } - } - } - - async save(): Promise { - if (!this._dirty) { - return; - } - const payload = JSON.stringify([...this._routes.values()], null, "\t"); - await lib.safeOutputFile(this._filePath, payload); - this._dirty = false; - } -} diff --git a/packages/controller/src/proxy_protocol.ts b/packages/controller/src/proxy_protocol.ts new file mode 100644 index 00000000..d166c622 --- /dev/null +++ b/packages/controller/src/proxy_protocol.ts @@ -0,0 +1,302 @@ +/** + * Encoding and decoding of the parts of the Factorio network protocol the + * player proxy needs to look at: the message header and the connection + * handshake messages. + * + * Based on the Factorio 2.x wire format. Every multi-byte value is little + * endian, "space optimized" values are a single byte unless that byte is 0xff, + * in which case the full width value follows it. + * + * @module + */ + +export const MessageType = { + Ping: 0, + PingReply: 1, + ConnectionRequest: 2, + ConnectionRequestReply: 3, + ConnectionRequestReplyConfirm: 4, + ConnectionAcceptOrDeny: 5, +} as const; + +const TYPE_MASK = 0x1f; +const FRAGMENTED_FLAG = 0x40; +const LAST_FRAGMENT_FLAG = 0x80; +const CONFIRM_FLAG = 0x8000; +const MESSAGE_ID_MASK = 0x7fff; + +/** Message types that carry a message id even when not fragmented */ +const RELIABLE_TYPES: ReadonlySet = new Set([ + MessageType.ConnectionRequest, + MessageType.ConnectionRequestReplyConfirm, +]); + +/** + * Max packet size Factorio servers tell clients to use, this is the largest + * datagram every host on the Internet is required to be able to receive. + */ +export const DEFAULT_MAX_PACKET_SIZE = 508; + +export interface ApplicationVersion { + major: number; + minor: number; + sub: number; + build: number; +} + +export function formatVersion(version: ApplicationVersion) { + return `${version.major}.${version.minor}.${version.sub}`; +} + +export function versionsEqual(a: ApplicationVersion, b: ApplicationVersion) { + return a.major === b.major && a.minor === b.minor && a.sub === b.sub && a.build === b.build; +} + +export interface MessageHeader { + type: number; + fragmented: boolean; + lastFragment: boolean; + messageId?: number; + fragmentNumber?: number; + confirms: number[]; + /** Offset of the payload of the message */ + payloadOffset: number; +} + +export interface ConnectionRequest { + header: MessageHeader; + version: ApplicationVersion; + clientRequestId: number; +} + +export interface ConnectionRequestReply { + version: ApplicationVersion; + clientRequestId: number; + serverRequestId: number; + maxPacketSize: number; +} + +export interface ConnectionRequestReplyConfirm { + clientRequestId: number; + serverRequestId: number; + /** Offset of serverRequestId in the packet */ + serverRequestIdOffset: number; + username: string; +} + +export class BufferReader { + constructor( + private _buffer: Buffer, + public offset = 0, + ) { } + + private _take(length: number) { + const start = this.offset; + if (start + length > this._buffer.length) { + throw new Error("Unexpected end of message"); + } + this.offset += length; + return start; + } + + readUInt8() { + return this._buffer.readUInt8(this._take(1)); + } + + readUInt16LE() { + return this._buffer.readUInt16LE(this._take(2)); + } + + readUInt32LE() { + return this._buffer.readUInt32LE(this._take(4)); + } + + /** Read a value stored as one byte or 0xff followed by a uint16 */ + readSpaceOptimizedUInt16() { + const value = this.readUInt8(); + return value === 0xff ? this.readUInt16LE() : value; + } + + /** Read a value stored as one byte or 0xff followed by a uint32 */ + readSpaceOptimizedUInt32() { + const value = this.readUInt8(); + return value === 0xff ? this.readUInt32LE() : value; + } + + readString() { + const length = this.readSpaceOptimizedUInt32(); + const start = this._take(length); + return this._buffer.toString("utf8", start, start + length); + } + + readVersion(): ApplicationVersion { + return { + major: this.readSpaceOptimizedUInt16(), + minor: this.readSpaceOptimizedUInt16(), + sub: this.readSpaceOptimizedUInt16(), + build: this.readUInt32LE(), + }; + } +} + +export class BufferWriter { + private _parts: number[] = []; + + writeUInt8(value: number) { + this._parts.push(value & 0xff); + } + + writeUInt16LE(value: number) { + this.writeUInt8(value); + this.writeUInt8(value >>> 8); + } + + writeUInt32LE(value: number) { + this.writeUInt16LE(value); + this.writeUInt16LE(value >>> 16); + } + + writeSpaceOptimizedUInt16(value: number) { + if (value >= 0xff) { + this.writeUInt8(0xff); + this.writeUInt16LE(value); + } else { + this.writeUInt8(value); + } + } + + writeVersion(version: ApplicationVersion) { + this.writeSpaceOptimizedUInt16(version.major); + this.writeSpaceOptimizedUInt16(version.minor); + this.writeSpaceOptimizedUInt16(version.sub); + this.writeUInt32LE(version.build); + } + + toBuffer() { + return Buffer.from(this._parts); + } +} + +export function messageType(buffer: Buffer) { + if (!buffer.length) { + return undefined; + } + return buffer[0] & TYPE_MASK; +} + +export function parseHeader(buffer: Buffer): MessageHeader { + const reader = new BufferReader(buffer); + const typeAndFlags = reader.readUInt8(); + const type = typeAndFlags & TYPE_MASK; + const fragmented = (typeAndFlags & FRAGMENTED_FLAG) !== 0; + const lastFragment = (typeAndFlags & LAST_FRAGMENT_FLAG) !== 0; + + let messageId: number | undefined; + let hasConfirm = false; + if (RELIABLE_TYPES.has(type) || fragmented) { + const messageIdAndFlags = reader.readUInt16LE(); + messageId = messageIdAndFlags & MESSAGE_ID_MASK; + hasConfirm = (messageIdAndFlags & CONFIRM_FLAG) !== 0; + } + + let fragmentNumber: number | undefined; + if (fragmented) { + fragmentNumber = reader.readSpaceOptimizedUInt16(); + } + + const confirms: number[] = []; + if (hasConfirm) { + const count = reader.readUInt8(); + for (let i = 0; i < count; i++) { + confirms.push(reader.readUInt32LE()); + } + } + + return { type, fragmented, lastFragment, messageId, fragmentNumber, confirms, payloadOffset: reader.offset }; +} + +export function parseConnectionRequest(buffer: Buffer): ConnectionRequest { + const header = parseHeader(buffer); + if (header.type !== MessageType.ConnectionRequest) { + throw new Error(`Expected ConnectionRequest, got message type ${header.type}`); + } + const reader = new BufferReader(buffer, header.payloadOffset); + const version = reader.readVersion(); + const clientRequestId = reader.readUInt32LE(); + return { header, version, clientRequestId }; +} + +export function parseConnectionRequestReply(buffer: Buffer): ConnectionRequestReply { + const header = parseHeader(buffer); + if (header.type !== MessageType.ConnectionRequestReply) { + throw new Error(`Expected ConnectionRequestReply, got message type ${header.type}`); + } + const reader = new BufferReader(buffer, header.payloadOffset); + return { + version: reader.readVersion(), + clientRequestId: reader.readUInt32LE(), + serverRequestId: reader.readUInt32LE(), + maxPacketSize: reader.readUInt16LE(), + }; +} + +/** + * Parse the start of a ConnectionRequestReplyConfirm + * + * The message contains the mod list and startup settings of the client and + * is usually fragmented, but the fields read here are always in the first + * fragment. + * + * @param buffer - Unfragmented message or first fragment of it. + * @returns the parsed fields. + */ +export function parseConnectionRequestReplyConfirm(buffer: Buffer): ConnectionRequestReplyConfirm { + const header = parseHeader(buffer); + if (header.type !== MessageType.ConnectionRequestReplyConfirm) { + throw new Error(`Expected ConnectionRequestReplyConfirm, got message type ${header.type}`); + } + if (header.fragmented && header.fragmentNumber !== 0) { + throw new Error(`Expected first fragment, got fragment ${header.fragmentNumber}`); + } + const reader = new BufferReader(buffer, header.payloadOffset); + const clientRequestId = reader.readUInt32LE(); + const serverRequestIdOffset = reader.offset; + const serverRequestId = reader.readUInt32LE(); + reader.readUInt32LE(); // instanceID + const username = reader.readString(); + return { clientRequestId, serverRequestId, serverRequestIdOffset, username }; +} + +export function encodeConnectionRequest(version: ApplicationVersion, clientRequestId: number, messageId = 0) { + const writer = new BufferWriter(); + writer.writeUInt8(MessageType.ConnectionRequest); + writer.writeUInt16LE(messageId & MESSAGE_ID_MASK); + writer.writeVersion(version); + writer.writeUInt32LE(clientRequestId); + return writer.toBuffer(); +} + +export function encodeConnectionRequestReply(reply: ConnectionRequestReply) { + const writer = new BufferWriter(); + writer.writeUInt8(MessageType.ConnectionRequestReply); + writer.writeVersion(reply.version); + writer.writeUInt32LE(reply.clientRequestId); + writer.writeUInt32LE(reply.serverRequestId); + writer.writeUInt16LE(reply.maxPacketSize); + return writer.toBuffer(); +} + +/** + * Turn a Ping into the PingReply a server would send back + * + * @param ping - Ping message received. + * @returns PingReply message, or undefined if ping was malformed. + */ +export function pingReplyFor(ping: Buffer) { + if (ping.length < 3) { + return undefined; + } + const reply = Buffer.from(ping.subarray(0, 3)); + reply[0] = MessageType.PingReply; + return reply; +} diff --git a/packages/host/src/Instance.ts b/packages/host/src/Instance.ts index 02bf3470..f70080d3 100644 --- a/packages/host/src/Instance.ts +++ b/packages/host/src/Instance.ts @@ -424,13 +424,6 @@ export default class Instance extends lib.Link { if (wasPaused) { this.hooks.autoPauseChanged.invoke(false); } - const hostId = this.config.get("instance.assigned_host"); - if (hostId !== null) { - this.sendTo( - "controller", - new lib.InstancePlayerRouteUpdateEvent(name, this.id, hostId, this.server.gamePort, Date.now()), - ); - } this.hooks.playerEvent.invoke(event); } diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 60a76be0..6de1ec56 100644 --- a/packages/lib/src/config/definitions.ts +++ b/packages/lib/src/config/definitions.ts @@ -247,15 +247,19 @@ export class ControllerConfig extends classes.Config { }, "controller.player_proxy_port": { title: "Player Proxy Port", - description: "UDP port to accept player connections for the controller proxy, set to null to disable.", + description: + "UDP port the controller accepts Factorio player connections on and forwards them to an " + + "instance, starting it if needed. Disabled when null.", restartRequired: true, type: "number", optional: true, - initialValue: 34197, + validator: validators.optional(validators.greaterThanZero), }, "controller.player_proxy_default_instance_id": { title: "Player Proxy Default Instance", - description: "Instance ID to route players to when no routing history is available.", + description: + "Instance ID to send players connecting through the player proxy to when no plugin picks one " + + "and they have not been on any instance before.", type: "number", optional: true, }, @@ -487,6 +491,7 @@ export interface InstanceConfigFields { "instance.id": number; "instance.assigned_host": number | null; "instance.auto_start": boolean; + "instance.player_proxy_port": number | null; "instance.exclude_from_start_all": boolean; [key: `${string}.load_plugin`]: boolean; [key: `${string}.version`]: string; @@ -561,6 +566,14 @@ export class InstanceConfig extends classes.Config { type: "boolean", initialValue: false, }, + "instance.player_proxy_port": { + description: + "Extra UDP port on the controller's player proxy that sends players straight to this instance, " + + "starting it if needed. Requires controller.player_proxy_port to be set.", + type: "number", + optional: true, + validator: validators.optional(validators.greaterThanZero), + }, "instance.exclude_from_start_all": { description: "Exclude this instance from the 'Start all' button operation", type: "boolean", diff --git a/packages/lib/src/data/messages_instance.ts b/packages/lib/src/data/messages_instance.ts index 2d10819c..d107632c 100644 --- a/packages/lib/src/data/messages_instance.ts +++ b/packages/lib/src/data/messages_instance.ts @@ -935,30 +935,3 @@ export class InstancePlayerUpdateEvent { return new this(json.type, json.name, PlayerStats.fromJSON(json.stats), json.reason); } } - -export class InstancePlayerRouteUpdateEvent { - declare ["constructor"]: typeof InstancePlayerRouteUpdateEvent; - static type = "event" as const; - static src = "instance" as const; - static dst = "controller" as const; - - constructor( - public name: string, - public instanceId: number, - public hostId: number, - public gamePort: number, - public timestampMs: number, - ) { } - - static jsonSchema = Type.Object({ - "name": Type.String(), - "instanceId": Type.Integer(), - "hostId": Type.Integer(), - "gamePort": Type.Integer(), - "timestampMs": Type.Number(), - }); - - static fromJSON(json: Static) { - return new this(json.name, json.instanceId, json.hostId, json.gamePort, json.timestampMs); - } -} diff --git a/packages/lib/src/link/messages.ts b/packages/lib/src/link/messages.ts index 06b08328..c4d95dcb 100644 --- a/packages/lib/src/link/messages.ts +++ b/packages/lib/src/link/messages.ts @@ -102,7 +102,6 @@ export const dataClasses: (RequestClass | EventClass) instance.InstanceAdminlistUpdateEvent, instance.InstanceWhitelistUpdateEvent, instance.InstancePlayerUpdateEvent, - instance.InstancePlayerRouteUpdateEvent, mod.ModPackGetRequest, mod.ModPackGetDefaultRequest, diff --git a/test/controller/PlayerProxyServer.test.js b/test/controller/PlayerProxyServer.test.js new file mode 100644 index 00000000..7b612e11 --- /dev/null +++ b/test/controller/PlayerProxyServer.test.js @@ -0,0 +1,376 @@ +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); + }); +});