diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index 1bbf7bef..e4aedcb6 100644 --- a/packages/lib/link/messages.js +++ b/packages/lib/link/messages.js @@ -368,7 +368,8 @@ messages.listInstances = new Request({ "id": { type: "integer" }, "assigned_slave": { type: ["null", "integer"] }, "status": { enum: [ - "unknown", "unassigned", "stopped", "starting", "running", "creating_save", "exporting_data", + "unknown", "unassigned", "stopped", "starting", "running", "stopping", + "creating_save", "exporting_data", ]}, }, }, @@ -747,7 +748,7 @@ messages.updateInstances = new Request({ properties: { "serialized_config": { type: "object" }, "status": { enum: [ - "stopped", "starting", "running", "creating_save", "exporting_data", + "stopped", "starting", "running", "stopping", "creating_save", "exporting_data", ]}, }, }, @@ -979,7 +980,7 @@ messages.instanceStatusChanged = new Event({ eventProperties: { "instance_id": { type: "integer" }, "status": { enum: [ - "stopped", "starting", "running", "creating_save", "exporting_data", + "stopped", "starting", "running", "stopping", "creating_save", "exporting_data", ]}, }, }); diff --git a/packages/lib/plugin.js b/packages/lib/plugin.js index fbae5fed..4dd71d08 100644 --- a/packages/lib/plugin.js +++ b/packages/lib/plugin.js @@ -191,7 +191,7 @@ class BaseInstancePlugin { * Plugins must stop sending messages to the master, or are forwarded * via the master server after the prepare disconnect has been handled. * - * @param {module:master/master~SlaveConnection} connection - + * @param {module:master/src/SlaveConnection} connection - * The connection to the slave preparing to disconnect. */ async onPrepareMasterDisconnect(connection) { } @@ -226,6 +226,11 @@ class BaseInstancePlugin { class BaseMasterPlugin { constructor(info, master, metrics, logger) { this.info = info; + + /** + * Master server + * @type {module:master/src/Master} + */ this.master = master; this.metrics = metrics; @@ -247,15 +252,19 @@ class BaseMasterPlugin { /** * Called when the status of an instance changes * - * Invoked when the master server has received notice from a slave that - * the running status of an instance has changed. The possible statuses - * that can be notified about are: - * - `stopped`: Instance is no longer running. + * Invoked when the master server has changed the status of an instance + * or received notice from a slave that the status of an instance has + * changed. The possible statuses that can be notified about are: + * - `unassigned:`: Instance is no longer asssigned to a slave. + * - `stopped`: Instance is no longer running or was just assigned to a + * slave. * - `starting`: Instance is in the process of starting up. * - `running`: Instance startup completed and is now running. + * - `stopping`: Instance is in the processing stopping. * - `creating_save`: Instance is in the process of creating a save. * - `exporting_data`: Instance is in the process of exporting item * icons and locale data. + * - `deleted`: Instance was deleted. * * On master startup all known instances gets the status `unknown` if * they are assigned to a slave, when the slave then connects to the @@ -263,15 +272,34 @@ class BaseMasterPlugin { * is invoked for all those instances. You can detect this situation by * checking if prev equals `unknown`. * + * When instances are created on the master they will notify of a status + * change with prev set to null. While the status of new instances is + * in most cases `unassigned` it's possible for the created instance to + * start with any state in some slave connection corner cases. + * * Note that it's possible for status change notification to get lost in * the case of network outages if reconnect fails to re-establish the * session between the master server and the slave. * * @param {Object} instance - the instance that changed. - * @param {string} prev - the previous status of the instance. + * @param {?string} prev - the previous status of the instance. */ async onInstanceStatusChanged(instance, prev) { } + /** + * Called when the value of an instance config field changed. + * + * Invoked after the value of the config field given by `field` has + * changed on an instance. + * + * @param {Object} instance - The instance the config changed on. + * @param {module:lib/config.ConfigGroup} group - + * The group who's field got changed on. + * @param {string} field - Name of the field that changed. + * @param {*} prev - The previous value of the field. + */ + async onInstanceConfigFieldChanged(instance, group, field, prev) { } + /** * Called before collecting Prometheus metrics * @@ -323,7 +351,7 @@ class BaseMasterPlugin { * * Invoked when the connection to the slave has been closed. * - * @param {module:master/master~SlaveConnection} connection - + * @param {module:master/src/SlaveConnection} connection - * The connection the event occured on. * @param {string} event - one of connect, drop, resume and close */ @@ -359,7 +387,7 @@ class BaseMasterPlugin { * * Invoked when the connection to the control has been closed. * - * @param {module:master/master~ControlConnection} connection - + * @param {module:master/src/ControlConnection} connection - * The connection the event occured on. * @param {string} event - one of connect, drop, resume, and close. */ @@ -375,7 +403,7 @@ class BaseMasterPlugin { * Plugins must stop sending messages to the slave in question after the * prepare disconnect has been handled. * - * @param {module:master/master~SlaveConnection} connection - + * @param {module:master/src/SlaveConnection} connection - * The connection to the slave preparing to disconnect. */ async onPrepareSlaveDisconnect(connection) { } @@ -404,7 +432,7 @@ class BaseMasterPlugin { * @param {Object} data - Data ta pass with the event. */ broadcastEventToSlaves(event, data={}) { - for (let slaveConnection of this.master.slaveConnections.values()) { + for (let slaveConnection of this.master.wsServer.slaveConnections.values()) { if ( !slaveConnection.connector.closing && (!event.plugin || slaveConnection.plugins.has(event.plugin)) diff --git a/packages/master/master.js b/packages/master/master.js index e70b488b..8c29a8a2 100755 --- a/packages/master/master.js +++ b/packages/master/master.js @@ -17,88 +17,28 @@ "use strict"; const path = require("path"); const fs = require("fs-extra"); -const jwt = require("jsonwebtoken"); const crypto = require("crypto"); const setBlocking = require("set-blocking"); -const events = require("events"); const yargs = require("yargs"); -const WebSocket = require("ws"); -const JSZip = require("jszip"); -const version = require("./package").version; const util = require("util"); const winston = require("winston"); -const http = require("http"); -const https = require("https"); - -// ugly globals -let masterConfig; -let masterConfigPath; -let masterPlugins = new Map(); -let pluginInfos = new Map(); -let stopAcceptingNewSessions = false; -let debugEvents = new events.EventEmitter(); -let loadedPlugins = {}; -let devMiddleware; -let clusterLogger; -let db = { - instances: null, - slaves: null, - roles: null, - users: null, -}; -let slaveConnections = new Map(); -let controlConnections = []; -let activeConnectors = new Map(); -let pendingSockets = new Set(); // homebrew modules -const HttpCloser = require("./src/HttpCloser"); -const routes = require("./src/routes"); -const libDatabase = require("@clusterio/lib/database"); -const libSchema = require("@clusterio/lib/schema"); -const libLink = require("@clusterio/lib/link"); const libErrors = require("@clusterio/lib/errors"); -const libPlugin = require("@clusterio/lib/plugin"); const libPluginLoader = require("@clusterio/lib/plugin_loader"); const libPrometheus = require("@clusterio/lib/prometheus"); const libConfig = require("@clusterio/lib/config"); const libUsers = require("@clusterio/lib/users"); const libSharedCommands = require("@clusterio/lib/shared_commands"); -const libHelpers = require("@clusterio/lib/helpers"); const { ConsoleTransport, levels, logger } = require("@clusterio/lib/logging"); const libLoggingUtils = require("@clusterio/lib/logging_utils"); -const express = require("express"); -const compression = require("compression"); -const cookieParser = require("cookie-parser"); -// Required for express post requests -const bodyParser = require("body-parser"); -const fileUpload = require("express-fileupload"); -const app = express(); -let httpServer; -let httpServerCloser; -let httpsServer; -let httpsServerCloser; +const Master = require("./src/Master"); +const UserManager = require("./src/UserManager"); +// globals +let master; -// Servers the web interface with the root path set apropriately. -function serveWeb(route) { - // The depth is is the number of slashes in the route minus one, but due - // to lenient matching on trailing slashes in the route we need to - // compensate if the request path contains a slash but not the route, - // and vice versa. - let routeDepth = (route.match(/\//g) || []).length - 1 - (route.slice(-1) === "/"); - return function(req, res, next) { - let depth = routeDepth + (req.path.slice(-1) === "/"); - let webRoot = "../".repeat(depth) || "./"; - fs.readFile(path.join(__dirname, "web", "index.html"), "utf8").then((content) => { - res.type("text/html"); - res.send(content.replace(/__WEB_ROOT__/g, webRoot)); - }).catch(err => { - next(err); - }); - }; -} const slaveMappingGauge = new libPrometheus.Gauge( "clusterio_master_slave_mapping", @@ -107,7 +47,10 @@ const slaveMappingGauge = new libPrometheus.Gauge( labels: ["slave_id", "slave_name"], callback: function() { slaveMappingGauge.clear(); - for (let [id, slave] of db.slaves) { + if (!master || !master.slaves) { + return; + } + for (let [id, slave] of master.slaves) { slaveMappingGauge.labels({ slave_id: String(id), slave_name: slave.name, @@ -124,7 +67,10 @@ const instanceMappingGauge = new libPrometheus.Gauge( labels: ["instance_id", "instance_name", "slave_id"], callback: function() { instanceMappingGauge.clear(); - for (let [id, instance] of db.instances) { + if (!master || !master.instances) { + return; + } + for (let [id, instance] of master.instances) { instanceMappingGauge.labels({ instance_id: String(id), instance_name: String(instance.config.get("instance.name")), @@ -135,1700 +81,33 @@ const instanceMappingGauge = new libPrometheus.Gauge( } ); -const hitCounter = new libPrometheus.Counter( - "clusterio_master_http_hits_total", - "How many HTTP requests in total have been received" -); - -const endpointHitCounter = new libPrometheus.Counter( - "clusterio_master_http_endpoint_hits_total", - "How many requests a particular HTTP endpoint has gotten", - { labels: ["route"] } -); - -const wsMessageCounter = new libPrometheus.Counter( - "clusterio_master_websocket_message_total", - "How many messages have been received over WebSocket on the master server", - { labels: ["direction"] } -); - -const wsConnectionsCounter = new libPrometheus.Counter( - "clusterio_master_websocket_connections_total", - "How many WebSocket connections have been initiated on the master server" -); - -const wsRejectedConnectionsCounter = new libPrometheus.Counter( - "clusterio_master_websocket_rejected_connections_total", - "How many WebSocket connections have been rejected during the handshake on the master server" -); - const wsActiveConnectionsGauge = new libPrometheus.Gauge( "clusterio_master_websocket_active_connections", - "How many WebSocket connections are currently open to the master server" + "How many WebSocket connections are currently open to the master server", + { callback: function(gauge) { gauge.set(master.wsServer.activeConnectors.size); }} ); const wsActiveSlavesGauge = new libPrometheus.Gauge( "clusterio_master_active_slaves", - "How many slaves are currently connected to the master" -); - - -app.use((req, res, next) => { - hitCounter.inc(); - next(); -}); -app.use(cookieParser()); -app.use(bodyParser.json({ - limit: "10mb", -})); -app.use(bodyParser.urlencoded({ - parameterLimit: 100000, - limit: "10mb", - extended: true, -})); -app.use(fileUpload()); -app.use(compression()); - -// Set folder to serve static content from (the website) -app.use(express.static(path.join(__dirname, "dist", "web"))); -app.use(express.static("static")); // Used for data export files - - -// Merges samples from sourceResult to destinationResult -function mergeSamples(destinationResult, sourceResult) { - // Merge metrics received by multiple slaves - let receivedSamples = new Map(sourceResult.samples); - for (let [suffix, suffixSamples] of destinationResult.samples) { - if (receivedSamples.has(suffix)) { - suffixSamples.push(...receivedSamples.get(suffix)); - receivedSamples.delete(suffix); - } - } - - for (let entry of receivedSamples) { - sourceResult.samples.push(entry); - } -} - -// Prometheus polling endpoint -async function getMetrics(req, res, next) { - endpointHitCounter.labels(req.route.path).inc(); - - let results = []; - let pluginResults = await libPlugin.invokeHook(masterPlugins, "onMetrics"); - for (let metricIterator of pluginResults) { - for await (let metric of metricIterator) { - results.push(metric); - } - } - - let requests = []; - let timeout = masterConfig.get("master.metrics_timeout") * 1000; - for (let slaveConnection of slaveConnections.values()) { - requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), timeout, null)); - } - - for await (let result of await libPrometheus.defaultRegistry.collect()) { - results.push(result); - } - - let resultMap = new Map(); - for (let response of await Promise.all(requests)) { - if (!response) { - // TODO: Log timeout occured? - continue; - } - - for (let result of response.results) { - if (!resultMap.has(result.metric.name)) { - resultMap.set(result.metric.name, result); - - } else { - // Merge metrics received by multiple slaves - mergeSamples(resultMap.get(result.metric.name), result); - } - } - } - - for (let result of resultMap.values()) { - results.push(libPrometheus.deserializeResult(result)); - } - - wsActiveConnectionsGauge.set(activeConnectors.size); - wsActiveSlavesGauge.set(slaveConnections.size); - - - let text = await libPrometheus.exposition(results); - res.set("Content-Type", libPrometheus.exposition.contentType); - res.send(text); -} -app.get("/metrics", (req, res, next) => getMetrics(req, res, next).catch(next)); - - -function validateSlaveToken(req, res, next) { - let token = req.header("x-access-token"); - if (!token) { - res.sendStatus(401); - return; - } - - try { - jwt.verify(token, masterConfig.get("master.auth_secret"), { audience: "slave" }); - - } catch (err) { - res.sendStatus(401); - return; - } - - next(); -} - -app.get("/api/plugins", (req, res) => { - let plugins = []; - for (let pluginInfo of pluginInfos) { - let name = pluginInfo.name; - let loaded = masterPlugins.has(name); - let enabled = loaded && masterConfig.group(pluginInfo.name).get("load_plugin"); - plugins.push({ name, version: pluginInfo.version, enabled, loaded }); - } - res.send(plugins); -}); - -// Handle an uploaded export package. -async function uploadExport(req, res, next) { - endpointHitCounter.labels(req.route.path).inc(); - if (req.get("Content-Type") !== "application/zip") { - res.sendStatus(415); - return; - } - - let data = []; - for await (let chunk of req) { - data.push(chunk); - } - data = Buffer.concat(data); - let zip = await JSZip.loadAsync(data); - data = null; - - // This is hardcoded to prevent path expansion attacks - let exportFiles = [ - "export/item-spritesheet.png", - "export/item-metadata.json", - "export/locale.json", - ]; - - for (let filePath of exportFiles) { - let file = zip.file(filePath); - if (!file) { - continue; - } - - let name = path.posix.basename(filePath); - await fs.outputFile(path.join("static", "export", name), await file.async("nodebuffer")); - } - - res.sendStatus(200); -} -app.put("/api/upload-export", - validateSlaveToken, - (req, res, next) => uploadExport(req, res, next).catch(next) + "How many slaves are currently connected to the master", + { callback: function(gauge) { gauge.set(master.wsServer.slaveConnections.size); }} ); const masterConnectedClientsCount = new libPrometheus.Gauge( "clusterio_master_connected_clients_count", "How many clients are currently connected to this master server", { labels: ["type"], callback: async function(gauge) { - gauge.labels("slave").set(slaveConnections.size); - gauge.labels("control").set(controlConnections.length); + gauge.labels("slave").set(master.wsServer.slaveConnections.size); + gauge.labels("control").set(master.wsServer.controlConnections.length); }, }, ); -/** - * Returns the URL needed to connect to the master server. - * - * @returns {string} master URL. - */ -function getMasterUrl() { - let url = masterConfig.get("master.external_address"); - if (!url) { - if (masterConfig.get("master.https_port")) { - url = `https://localhost:${masterConfig.get("master.https_port")}/`; - } else { - url = `http://localhost:${masterConfig.get("master.http_port")}/`; - } - } - return url; -} -/** - * Load Map from JSON file in the database directory. - * - * @param {string} databaseDirectory - Path to master database directory. - * @param {string} file - Name of file to load. - * @returns {Map} file loaded as a map. - */ -async function loadMap(databaseDirectory, file) { - let databasePath = path.resolve(databaseDirectory, file); - logger.info(`Loading ${databasePath}`); - return await libDatabase.loadJsonArrayAsMap(databasePath); -} - -async function loadInstances(databaseDirectory, file) { - let filePath = path.join(databaseDirectory, file); - logger.info(`Loading ${filePath}`); - - let instances = new Map(); - try { - let serialized = JSON.parse(await fs.readFile(filePath)); - for (let serializedConfig of serialized) { - let instanceConfig = new libConfig.InstanceConfig("master"); - await instanceConfig.load(serializedConfig); - let status = instanceConfig.get("instance.assigned_slave") === null ? "unassigned" : "unknown"; - instances.set(instanceConfig.get("instance.id"), { config: instanceConfig, status }); - } - - } catch (err) { - if (err.code !== "ENOENT") { - throw err; - } - } - - return instances; -} - -async function loadUsers(databaseDirectory, file) { - let loadedRoles = new Map(); - let loadedUsers = new Map(); - try { - let content = JSON.parse(await fs.readFile(path.join(databaseDirectory, file))); - for (let serializedRole of content.roles) { - let role = new libUsers.Role(serializedRole); - loadedRoles.set(role.id, role); - } - - for (let serializedUser of content.users) { - let user = new libUsers.User(serializedUser, loadedRoles); - loadedUsers.set(user.name, user); - } - - } catch (err) { - if (err.code !== "ENOENT") { - throw err; - } - - // Create default roles if loading failed - libUsers.ensureDefaultAdminRole(loadedRoles); - libUsers.ensureDefaultPlayerRole(loadedRoles); - } - - db.roles = loadedRoles; - db.users = loadedUsers; -} - -/** - * Create a new user - * - * Creates a new user instances and add it to the user database. - * - * @param {string} name - Name of the user to create. - * @returns {module:lib/users.User} newly created user. - */ -function createUser(name) { - if (db.users.has(name)) { - throw new Error(`User '${name}' already exists`); - } - - let defaultRoleId = masterConfig.get("master.default_role_id"); - let user = new libUsers.User({ name, roles: [defaultRoleId] }, db.roles); - db.users.set(name, user); - return user; -} - -/** - * Save Map to JSON file in the database directory. - * - * @param {string} databaseDirectory - Path to master database directory. - * @param {string} file - Name of file to save. - * @param {Map} map - Mapping to save into file. - */ -async function saveMap(databaseDirectory, file, map) { - let databasePath = path.resolve(databaseDirectory, file); - logger.info(`Saving ${databasePath}`); - await libDatabase.saveMapAsJsonArray(databasePath, map); -} - -async function saveInstances(databaseDirectory, file, instances) { - let filePath = path.join(databaseDirectory, file); - let serialized = []; - for (let instance of instances.values()) { - serialized.push(instance.config.serialize()); - } - - await fs.outputFile(filePath, JSON.stringify(serialized, null, 4)); -} - -async function saveUsers(databaseDirectory, file) { - let filePath = path.join(databaseDirectory, file); - let serializedRoles = []; - for (let role of db.roles.values()) { - serializedRoles.push(role.serialize()); - } - - let serializedUsers = []; - for (let user of db.users.values()) { - serializedUsers.push(user.serialize()); - } - - let serialized = { - users: serializedUsers, - roles: serializedRoles, - }; - await fs.outputFile(filePath, JSON.stringify(serialized, null, 4)); -} - -let shuttingDown = false; -/** - * Innitiate shutdown of master server - */ -async function shutdown() { - if (shuttingDown) { - return; - } - shuttingDown = true; - - let exitStartTime = Date.now(); - try { - logger.info("Saving configs"); - await fs.outputFile(masterConfigPath, JSON.stringify(masterConfig.serialize(), null, 4)); - - if (devMiddleware) { - await new Promise((resolve, reject) => { devMiddleware.close(resolve); }); - } - - if (!db.slaves) { - return; - } - - await saveMap(masterConfig.get("master.database_directory"), "slaves.json", db.slaves); - await saveInstances(masterConfig.get("master.database_directory"), "instances.json", db.instances); - await saveUsers(masterConfig.get("master.database_directory"), "users.json"); - - await libPlugin.invokeHook(masterPlugins, "onShutdown"); - - stopAcceptingNewSessions = true; - - let disconnectTasks = []; - for (let controlConnection of controlConnections) { - controlConnection.connector.setTimeout(masterConfig.get("master.connector_shutdown_timeout")); - disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit")); - } - - for (let slaveConnection of slaveConnections.values()) { - slaveConnection.connector.setTimeout(masterConfig.get("master.connector_shutdown_timeout")); - disconnectTasks.push(slaveConnection.disconnect(1001, "Server Quit")); - } - - logger.info(`Waiting for ${disconnectTasks.length} connectors to close`); - for (let task of disconnectTasks) { - try { - await task; - } catch (err) { - if (!(err instanceof libErrors.SessionLost)) { - logger.warn(`Unexpected error disconnecting connector:\n${err.stack}`); - } - } - } - - for (let socket of pendingSockets) { - socket.close(1001, "Server Quit"); - } - - let stopTasks = []; - logger.info("Stopping HTTP(S) server"); - if (httpServer) { stopTasks.push(httpServerCloser.close()); } - if (httpsServer) { stopTasks.push(httpsServerCloser.close()); } - await Promise.all(stopTasks); - - logger.info(`Clusterio cleanly exited in ${Date.now() - exitStartTime}ms`); - - } catch (err) { - setBlocking(true); - logger.fatal(` -+--------------------------------------------------------------------+ -| Unexpected error occured while shutting down master, please report | -| it to https://github.com/clusterio/factorioClusterio/issues | -+--------------------------------------------------------------------+ -${err.stack}` - ); - // eslint-disable-next-line no-process-exit - process.exit(1); - } -} - -/** - * Base class for master server connections - * - * @extends module:lib/link.Link - */ -class BaseConnection extends libLink.Link { - constructor(target, connector) { - super("master", target, connector); - libLink.attachAllMessages(this); - for (let masterPlugin of masterPlugins.values()) { - libPlugin.attachPluginMessages(this, masterPlugin); - } - } - - async forwardRequestToInstance(message, request) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); - } - - let slaveId = instance.config.get("instance.assigned_slave"); - if (slaveId === null) { - throw new libErrors.RequestError("Instance is not assigned to a slave"); - } - - let connection = slaveConnections.get(slaveId); - if (!connection) { - throw new libErrors.RequestError("Slave containing instance is not connected"); - } - if (request.plugin && !connection.plugins.has(request.plugin)) { - throw new libErrors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`); - } - - return await request.send(connection, message.data); - } - - async forwardEventToInstance(message, event) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { return; } - - let slaveId = instance.config.get("instance.assigned_slave"); - if (slaveId === null) { return; } - - let connection = slaveConnections.get(slaveId); - if (!connection || connection.closing) { return; } - if (event.plugin && !connection.plugins.has(event.plugin)) { return; } - - event.send(connection, message.data); - } - - async broadcastEventToSlaves(message, event) { - for (let slaveConnection of slaveConnections.values()) { - // Do not broadcast back to the source - if (slaveConnection === this) { continue; } - if (slaveConnection.connector.closing) { continue; } - if (event.plugin && !slaveConnection.plugins.has(event.plugin)) { continue; } - - event.send(slaveConnection, message.data); - } - } - - async broadcastEventToInstance(message, event) { - await this.broadcastEventToSlaves(message, event); - } - - async prepareDisconnectRequestHandler(message, request) { - await libPlugin.invokeHook(masterPlugins, "onPrepareSlaveDisconnect", this); - this.connector.setClosing(); - return await super.prepareDisconnectRequestHandler(message, request); - } - - async disconnect(code, reason) { - try { - await libLink.messages.prepareDisconnect.send(this); - } catch (err) { - if (!(err instanceof libErrors.SessionLost)) { - logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`); - } - } - - await this.connector.close(code, reason); - } -} - -const lastQueryLogTime = new libPrometheus.Gauge( - "clusterio_master_last_query_log_duration_seconds", - "Time in seconds the last log query took to execute." -); - -class ControlConnection extends BaseConnection { - constructor(registerData, connector, user) { - super("control", connector); - - this._agent = registerData.agent; - this._version = registerData.version; - - /** - * The user making this connection. - * @type {module:lib/user.User} - */ - this.user = user; - - this.connector.on("close", () => { - let index = controlConnections.indexOf(this); - if (index !== -1) { - controlConnections.splice(index, 1); - } - }); - - this.logTransport = null; - this.logSubscriptions = { - all: false, - master: false, - slave_ids: [], - instance_ids: [], - }; - - this.ws_dumper = null; - this.connector.on("connect", () => { - this.connector._socket.clusterio_ignore_dump = Boolean(this.ws_dumper); - }); - this.connector.on("close", () => { - if (this.logTransport) { - this.logTransport = null; - logger.remove(this.logTransport); - } - if (this.ws_dumper) { - debugEvents.off("message", this.ws_dumper); - } - }); - - for (let event of ["connect", "drop", "resume", "close"]) { - this.connector.on(event, () => { - for (let masterPlugin of masterPlugins.values()) { - masterPlugin.onControlConnectionEvent(this, event); - } - }); - } - } - - async getMasterConfigRequestHandler() { - return { serialized_config: masterConfig.serialize("control") }; - } - - async setMasterConfigFieldRequestHandler(message) { - masterConfig.set(message.data.field, message.data.value, "control"); - } - - async setMasterConfigPropRequestHandler(message) { - let { field, prop, value } = message.data; - masterConfig.setProp(field, prop, value, "control"); - } - - async listSlavesRequestHandler(message) { - let list = []; - for (let slave of db.slaves.values()) { - list.push({ - agent: slave.agent, - version: slave.version, - id: slave.id, - name: slave.name, - connected: slaveConnections.has(slave.id), - }); - } - return { list }; - } - - generateSlaveToken(slaveId) { - return jwt.sign({ aud: "slave", slave: slaveId }, masterConfig.get("master.auth_secret")); - } - - async generateSlaveTokenRequestHandler(message) { - return { token: this.generateSlaveToken(message.data.slave_id) }; - } - - async createSlaveConfigRequestHandler(message) { - let slaveConfig = new libConfig.SlaveConfig("control"); - await slaveConfig.init(); - - slaveConfig.set("slave.master_url", getMasterUrl()); - if (message.data.id !== null) { - slaveConfig.set("slave.id", message.data.id); - } - if (message.data.name !== null) { - slaveConfig.set("slave.name", message.data.name); - } - if (message.data.generate_token) { - this.user.checkPermission("core.slave.generate_token"); - slaveConfig.set("slave.master_token", this.generateSlaveToken(slaveConfig.get("slave.id"))); - } - return { serialized_config: slaveConfig.serialize() }; - } - - async listInstancesRequestHandler(message) { - let list = []; - for (let instance of db.instances.values()) { - list.push({ - id: instance.config.get("instance.id"), - name: instance.config.get("instance.name"), - assigned_slave: instance.config.get("instance.assigned_slave"), - status: instance.status, - }); - } - return { list }; - } - - // XXX should probably add a hook for slave reuqests? - async createInstanceRequestHandler(message) { - let instanceConfig = new libConfig.InstanceConfig("master"); - await instanceConfig.load(message.data.serialized_config); - - let instanceId = instanceConfig.get("instance.id"); - if (db.instances.has(instanceId)) { - throw new libErrors.RequestError(`Instance with ID ${instanceId} already exists`); - } - - // Add common settings for the Factorio server - let settings = { - "name": `${masterConfig.get("master.name")} - ${instanceConfig.get("instance.name")}`, - "description": `Clusterio instance for ${masterConfig.get("master.name")}`, - "tags": ["clusterio"], - "max_players": 0, - "visibility": { "public": true, "lan": true }, - "username": "", - "token": "", - "game_password": "", - "require_user_verification": true, - "max_upload_in_kilobytes_per_second": 0, - "max_upload_slots": 5, - "ignore_player_limit_for_returning_players": false, - "allow_commands": "admins-only", - "autosave_interval": 10, - "autosave_slots": 5, - "afk_autokick_interval": 0, - "auto_pause": false, - "only_admins_can_pause_the_game": true, - "autosave_only_on_server": true, - - ...instanceConfig.get("factorio.settings"), - }; - instanceConfig.set("factorio.settings", settings); - - db.instances.set(instanceId, { config: instanceConfig, status: "unassigned" }); - } - - async deleteInstanceRequestHandler(message, request) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); - } - - if (instance.config.get("instance.assigned_slave") !== null) { - await this.forwardRequestToInstance(message, request); - } - db.instances.delete(message.data.instance_id); - } - - async getInstanceConfigRequestHandler(message) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); - } - - return { - serialized_config: instance.config.serialize("control"), - }; - } - - async updateInstanceConfig(instance) { - let slaveId = instance.config.get("instance.assigned_slave"); - if (slaveId) { - let connection = slaveConnections.get(slaveId); - if (connection) { - await libLink.messages.assignInstance.send(connection, { - instance_id: instance.config.get("instance.id"), - serialized_config: instance.config.serialize("slave"), - }); - } - } - } - - async setInstanceConfigFieldRequestHandler(message) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); - } - - if (message.data.field === "instance.assigned_slave") { - throw new libErrors.RequestError("instance.assigned_slave must be set through the assign-slave interface"); - } - - if (message.data.field === "instance.id") { - // XXX is this worth implementing? It's race condition galore. - throw new libErrors.RequestError("Setting instance.id is not supported"); - } - - instance.config.set(message.data.field, message.data.value, "control"); - await this.updateInstanceConfig(instance); - } - - async setInstanceConfigPropRequestHandler(message) { - let instance = db.instances.get(message.data.instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); - } - - let { field, prop, value } = message.data; - instance.config.setProp(field, prop, value, "control"); - await this.updateInstanceConfig(instance); - } - - async assignInstanceCommandRequestHandler(message, request) { - let { slave_id, instance_id } = message.data; - let instance = db.instances.get(instance_id); - if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`); - } - - // Check if target slave is connected - let newSlaveConnection; - if (slave_id !== null) { - newSlaveConnection = slaveConnections.get(slave_id); - if (!newSlaveConnection) { - // The case of the slave not getting the assign instance message - // still have to be handled, so it's not a requirement that the - // target slave be connected to the master while doing the - // assignment, but it is IMHO a better user experience if this - // is the case. - throw new libErrors.RequestError("Target slave is not connected to the master server"); - } - } - - // Unassign from currently assigned slave if it is connected. - let currentAssignedSlave = instance.config.get("instance.assigned_slave"); - if (currentAssignedSlave !== null && slave_id !== currentAssignedSlave) { - let oldSlaveConnection = slaveConnections.get(currentAssignedSlave); - if (oldSlaveConnection && !oldSlaveConnection.connector.closing) { - await libLink.messages.unassignInstance.send(oldSlaveConnection, { instance_id }); - } - } - - // Assign to target - instance.config.set("instance.assigned_slave", slave_id); - if (slave_id !== null) { - await libLink.messages.assignInstance.send(newSlaveConnection, { - instance_id, - serialized_config: instance.config.serialize("slave"), - }); - } - } - - async setLogSubscriptionsRequestHandler(message) { - this.logSubscriptions = message.data; - this.updateLogSubscriptions(); - } - - static logFilter({ all, master, slave_ids, instance_ids, max_level }) { - return info => { - // Note: reversed to filter out undefined levels - if (max_level && !(levels[info.level] <= levels[max_level])) { - return false; - } - - if (all) { - return true; - } - if (master && info.slave_id === undefined) { - return true; - } - if (info.slave_id && slave_ids.includes(info.slave_id)) { - return true; - } - if (info.instance_id && instance_ids.includes(info.instance_id)) { - return true; - } - return false; - }; - } - - updateLogSubscriptions() { - let { all, master, slave_ids, instance_ids } = this.logSubscriptions; - if (all || master || slave_ids.length || instance_ids.length) { - if (!this.logTransport) { - this.logTransport = new libLoggingUtils.LinkTransport({ link: this }); - clusterLogger.add(this.logTransport); - } - this.logTransport.filter = this.constructor.logFilter(this.logSubscriptions); - - } else if (this.logTransport) { - clusterLogger.remove(this.logTransport); - this.logTransport = null; - } - } - - async queryLogRequestHandler(message) { - let transport = new winston.transports.File({ filename: "cluster.log" }); - let query = util.promisify((options, callback) => transport.query(options, callback)); - - // Available options and defaults inferred from reading the source - // rows: number = limit || 10 - // limit: number // alias of rows. - // start: number = 0 - // until: Date = now - // from: Date = until - 24h - // order: "asc"|"desc" = "desc" - // fields: Array - // level: string - // This interface is junk: - // - The level option filters by exact match - // - No way to add your own filter - // - No way to stream results - // - The log file is read starting from the beginning - let setDuration = lastQueryLogTime.startTimer(); - let log = await query({ - order: "asc", - limit: Infinity, - from: new Date(0), - }); - - log = log.filter(this.constructor.logFilter(message.data)); - setDuration(); - return { log }; - } - - async listPermissionsRequestHandler(message) { - let list = []; - for (let permission of libUsers.permissions.values()) { - list.push({ - name: permission.name, - title: permission.title, - description: permission.description, - }); - } - return { list }; - } - - async listRolesRequestHandler(message) { - let list = []; - for (let role of db.roles.values()) { - list.push({ - id: role.id, - name: role.name, - description: role.description, - permissions: [...role.permissions], - }); - } - return { list }; - } - - async createRoleRequestHandler(message) { - let lastId = Math.max.apply(null, [...db.roles.keys()]); - - // Start at 5 to leave space for future default roles - let id = Math.max(5, lastId+1); - db.roles.set(id, new libUsers.Role({ id, ...message.data })); - return { id }; - } - - async updateRoleRequestHandler(message) { - let { id, name, description, permissions } = message.data; - let role = db.roles.get(id); - if (!role) { - throw new libErrors.RequestError(`Role with ID ${id} does not exist`); - } - - role.name = name; - role.description = description; - role.permissions = new Set(permissions); - } - - async grantDefaultRolePermissionsRequestHandler(message) { - let role = db.roles.get(message.data.id); - if (!role) { - throw new libErrors.RequestError(`Role with ID ${message.data.id} does not exist`); - } - - role.grantDefaultPermissions(); - } - - async deleteRoleRequestHandler(message) { - let id = message.data.id; - let role = db.roles.get(id); - if (!role) { - throw new libErrors.RequestError(`Role with ID ${id} does not exist`); - } - - db.roles.delete(id); - for (let user of db.users.values()) { - user.roles.delete(role); - } - } - - async listUsersRequestHandler(message) { - let list = []; - for (let user of db.users.values()) { - list.push({ - name: user.name, - roles: [...user.roles].map(role => role.id), - is_admin: user.isAdmin, - is_banned: user.isBanned, - is_whitelisted: user.isWhitelisted, - instances: [...user.instances], - }); - } - return { list }; - } - - async createUserRequestHandler(message) { - createUser(message.data.name); - } - - async updateUserRolesRequestHandler(message) { - let user = db.users.get(message.data.name); - if (!user) { - throw new libErrors.RequestError(`User '${message.data.name}' does not exist`); - } - - let resolvedRoles = new Set(); - for (let roleId of message.data.roles) { - let role = db.roles.get(roleId); - if (!role) { - throw new libErrors.RequestError(`Role with ID ${roleId} does not exist`); - } - - resolvedRoles.add(role); - } - - user.roles = resolvedRoles; - } - - async setUserAdminRequestHandler(message) { - let { name, create, admin } = message.data; - let user = db.users.get(name); - if (!user) { - if (create) { - this.user.checkPermission("core.user.create"); - user = createUser(name); - } else { - throw new libErrors.RequestError(`User '${name}' does not exist`); - } - } - - user.isAdmin = admin; - this.broadcastEventToSlaves({ data: { name, admin }}, libLink.messages.adminlistUpdate); - } - - async setUserBannedRequestHandler(message) { - let { name, create, banned, reason } = message.data; - let user = db.users.get(name); - if (!user) { - if (create) { - this.user.checkPermission("core.user.create"); - user = createUser(name); - } else { - throw new libErrors.RequestError(`User '${name}' does not exist`); - } - } - - user.isBanned = banned; - user.banReason = reason; - this.broadcastEventToSlaves({ data: { name, banned, reason }}, libLink.messages.banlistUpdate); - } - - async setUserWhitelistedRequestHandler(message) { - let { name, create, whitelisted } = message.data; - let user = db.users.get(name); - if (!user) { - if (create) { - this.user.checkPermission("core.user.create"); - user = createUser(name); - } else { - throw new libErrors.RequestError(`User '${name}' does not exist`); - } - } - - user.isWhitelisted = whitelisted; - this.broadcastEventToSlaves({ data: { name, whitelisted }}, libLink.messages.whitelistUpdate); - } - - async deleteUserRequestHandler(message) { - let user = db.users.get(message.data.name); - if (!user) { - throw new libErrors.RequestError(`User '${message.data.name}' does not exist`); - } - - if (user.is_admin) { - this.broadcastEventToSlaves({ data: { name, admin: false }}, libLink.messages.adminlistUpdate); - } - if (user.is_whitelisted) { - this.broadcastEventToSlaves({ data: { name, whitelisted: false }}, libLink.messages.whitelistUpdate); - } - if (user.is_banned) { - this.broadcastEventToSlaves({ data: { name, banned: false, reason: "" }}, libLink.messages.banlistUpdate); - } - db.users.delete(message.data.name); - } - - async debugDumpWsRequestHandler(message) { - this.ws_dumper = data => { - if (this.connector.connected) { - libLink.messages.debugWsMessage.send(this, data); - } - }; - this.connector._socket.clusterio_ignore_dump = true; - debugEvents.on("message", this.ws_dumper); - } -} - -/** - * Represents the connection to a slave - * - * @extends module:master~BaseConnection - */ -class SlaveConnection extends BaseConnection { - constructor(registerData, connector) { - super("slave", connector); - - this._agent = registerData.agent; - this._id = registerData.id; - this._name = registerData.name; - this._version = registerData.version; - this.plugins = new Map(Object.entries(registerData.plugins)); - - db.slaves.set(this._id, { - agent: this._agent, - id: this._id, - name: this._name, - version: this._version, - plugins: registerData.plugins, - }); - - this.connector.on("close", () => { - if (slaveConnections.get(this._id) === this) { - slaveConnections.delete(this._id); - } - }); - - for (let event of ["connect", "drop", "resume", "close"]) { - // eslint-disable-next-line no-loop-func - this.connector.on(event, () => { - for (let masterPlugin of masterPlugins.values()) { - masterPlugin.onSlaveConnectionEvent(this, event); - } - }); - } - } - - async instanceStatusChangedEventHandler(message, event) { - let instance = db.instances.get(message.data.instance_id); - - // It's possible to get updates from an instance that does not exist - // or is not assigned to the slave it originated from if it was - // reassigned or deleted while the connection to the slave it was - // originally on was down at the time. - if (!instance || instance.config.get("instance.assigned_slave") !== this._id) { - logger.warn(`Got bogus update for instance id ${message.data.instance_id}`); - return; - } - - let prev = instance.status; - instance.status = message.data.status; - logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`); - await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); - } - - async updateInstancesRequestHandler(message) { - // Push updated instance configs - for (let instance of db.instances.values()) { - if (instance.config.get("instance.assigned_slave") === this._id) { - await libLink.messages.assignInstance.send(this, { - instance_id: instance.config.get("instance.id"), - serialized_config: instance.config.serialize("slave"), - }); - } - } - - // Assign instances the slave has but master does not - for (let instance of message.data.instances) { - let instanceConfig = new libConfig.InstanceConfig("master"); - await instanceConfig.load(instance.serialized_config, "slave"); - - let masterInstance = db.instances.get(instanceConfig.get("instance.id")); - if (masterInstance) { - // Check if this instance is assigned somewhere else. - if (masterInstance.config.get("instance.assigned_slave") !== this._id) { - await libLink.messages.unassignInstance.send(this, { - instance_id: masterInstance.config.get("instance.id"), - }); - continue; - } - - // Already have this instance, update state instead - if (masterInstance.status !== instance.status) { - let prev = masterInstance.status; - masterInstance.status = instance.status; - logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instance.status}`); - await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); - } - continue; - } - - instanceConfig.set("instance.assigned_slave", this._id); - db.instances.set(instanceConfig.get("instance.id"), { - config: instanceConfig, - status: instance.status, - }); - await libLink.messages.assignInstance.send(this, { - instance_id: instanceConfig.get("instance.id"), - serialized_config: instanceConfig.serialize("slave"), - }); - } - - // Push lists to make sure they are in sync. - let adminlist = []; - let banlist = []; - let whitelist = []; - - for (let user of db.users.values()) { - if (user.isAdmin) { - adminlist.push(user.name); - } - if (user.isBanned) { - banlist.push([user.name, user.banReason]); - } - if (user.isWhitelisted) { - whitelist.push(user.name); - } - } - - libLink.messages.syncUserLists.send(this, { adminlist, banlist, whitelist }); - } - - async logMessageEventHandler(message) { - clusterLogger.log({ - ...message.data.info, - slave_id: this._id, - slave_name: this._name, - }); - } - - async playerEventEventHandler(message) { - let { instance_id, name, type } = message.data; - let user = db.users.get(name); - if (!user) { - user = createUser(name); - } - - if (type === "join") { - user.notifyJoin(instance_id); - } else if (type === "leave") { - user.notifyLeave(instance_id); - } - - let instance = db.instances.get(instance_id); - await libPlugin.invokeHook(masterPlugins, "onPlayerEvent", instance, message.data); - } -} - -const wss = new WebSocket.Server({ - noServer: true, - path: "/api/socket", -}); - -/** - * Returns true if value is a signed 32-bit integer - * - * @param {number} value - value to test. - * @returns {Boolean} - * true if value is an integer between -231 and - * 231-1. - */ -function isInteger(value) { - return (value | 0) === value; -} - - -/** - * Connector for master server connections - * - * @extends module:lib/link.WebSocketBaseConnector - */ -class WebSocketServerConnector extends libLink.WebSocketBaseConnector { - constructor(socket, sessionId) { - super(); - - this._socket = socket; - this._sessionId = sessionId; - - // The following states are used in the server connector - // handshake: Waiting for client to (re)connect. - // connected: Connection is online - // closing: Connection is in the process of being closed. - // closed: Connection has been closed - this._state = "handshake"; - this._connected = false; - this._timeout = 15 * 60; - } - - _check(expectedStates) { - if (!expectedStates.includes(this._state)) { - throw new Error( - `Expected state ${expectedStates} but state is ${this._state}` - ); - } - } - - /** - * Send ready over the socket - * - * Sends the ready message over the socket to initiate the session. - * - * @param {string} sessionToken - - * the session token to send to the client. - * @param {Object=} additionalData - - * extra properties to send along the ready message. - */ - ready(sessionToken, additionalData) { - this._heartbeatInterval = masterConfig.get("master.heartbeat_interval"); - this._socket.send(JSON.stringify({ - seq: null, - type: "ready", - data: { - session_token: sessionToken, - heartbeat_interval: this._heartbeatInterval, - ...additionalData, - }, - })); - - this._state = "connected"; - this._connected = true; - this._attachSocketHandlers(); - this.emit("connect"); - } - - /** - * Continue connection with the given socket - * - * Terminates the current socket and contiunes the session over the - * socket given from the message sequence given. - * - * @param {module:net.Socket} socket - New socket to continue on. - * @param {number} lastSeq - The last message the client received. - */ - continue(socket, lastSeq) { - - // It's possible the previous connection hasn't closed yet due to a - // stale connection. If this is the case the connector will have to - // wait for the previous socket to close down before it can continue - // from a new socket, or there will be overlapping events from both. - if (["connected", "closing"].includes(this._state)) { - this._socket.close(4003, "Session Hijacked"); - - // Correctly waiting for the connector to close and then - // handling all the edge cases like multiple continues happening - // in parallel or connections going stale is far too difficult. - // Kill the connection here and let the client reconnect later - // when the connector is not held up by an active connection. - socket.close(1013, "Session Busy"); - socket.terminate(); - return; - } - - this._socket = socket; - - if (this._timeoutId) { - clearTimeout(this._timeoutId); - this._timeoutId = null; - } - - this._heartbeatInterval = masterConfig.get("master.heartbeat_interval"); - this._socket.send(JSON.stringify({ - seq: null, - type: "continue", - data: { - last_seq: this._lastReceivedSeq, - heartbeat_interval: this._heartbeatInterval, - }, - })); - - this._state = "connected"; - this._connected = true; - this._attachSocketHandlers(); - this._dropSendBufferSeq(lastSeq); - for (let message of this._sendBuffer) { - this._socket.send(JSON.stringify(message)); - } - this.emit("resume"); - } - - setTimeout(timeout) { - if (this._timeoutId) { - clearTimeout(this._timeoutId); - } - - this._timeoutId = setTimeout(() => { this._timedOut(); }, timeout * 1000); - this._timeout = timeout; - } - - _timedOut() { - logger.verbose("SOCKET | Connection timed out"); - this._timeoutId = null; - this._lastReceivedSeq = null; - this._sendBuffer.length = 0; - this._state = "closed"; - activeConnectors.delete(this._sessionId); - this.emit("close"); - this.emit("invalidate"); - } - - _attachSocketHandlers() { - this.startHeartbeat(); - - this._socket.on("close", (code, reason) => { - logger.verbose(`SOCKET | Close (code: ${code}, reason: ${reason})`); - if (this._state === "closing") { - this._lastReceivedSeq = null; - this._sendBuffer.length = 0; - this._state = "closed"; - activeConnectors.delete(this._sessionId); - - if (this._connected) { - this._connected = false; - this.emit("close"); - } - - if (this._timeoutId) { - clearTimeout(this._timeoutId); - this._timeoutId = null; - } - - } else { - this._state = "handshake"; - this.emit("drop"); - this._timeoutId = setTimeout(() => { this._timedOut(); }, this._timeout * 1000); - } - - this.stopHeartbeat(); - }); - - this._socket.on("error", err => { - // It's assumed that close is always called by ws - logger.verbose("SOCKET | Error:", err); - }); - - this._socket.on("open", () => { - logger.verbose("SOCKET | Open"); - }); - this._socket.on("ping", data => { - logger.verbose(`SOCKET | Ping (data: ${data}`); - }); - this._socket.on("pong", data => { - logger.verbose(`SOCKET | Pong (data: ${data}`); - }); - - // Handle messages - this._socket.on("message", data => { - let message = JSON.parse(data); - if (["connected", "closing"].includes(this._state)) { - if (message.seq !== null) { - this._lastReceivedSeq = message.seq; - } - - if (message.type === "heartbeat") { - this._processHeartbeat(message); - - } else { - this.emit("message", message); - } - - } else { - throw new Error(`Received message in unexpected state ${this._state}`); - } - }); - } - - /** - * Close the connection with the given reason. - * - * Sends a close frame and disconnects the connector. - * - * @param {number} code - WebSocket close code. - * @param {string} reason - WebSocket close reason. - */ - async close(code, reason) { - if (this._state === "closed") { - return; - } - - this.stopHeartbeat(); - this._state = "closing"; - this._socket.close(code, reason); - await events.once(this, "close"); - } -} - -let nextSessionId = 1; - -// Unique string for the session token audience -let masterSession = `session-${Date.now()}`; - -// Handle socket handshake -async function handleHandshake(message, socket, req, attachHandler) { - try { - message = JSON.parse(message); - } catch (err) { - logger.verbose(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid JSON`); - wsRejectedConnectionsCounter.inc(); - socket.close(1002, "Invalid JSON"); - return; - } - - if (!libSchema.clientHandshake(message)) { - logger.verbose(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid handshake`); - wsRejectedConnectionsCounter.inc(); - socket.close(1002, "Bad handshake"); - return; - } - - let { seq, type, data } = message; - - if (type === "resume") { - let connector; - try { - let payload = jwt.verify( - data.session_token, - masterConfig.get("master.auth_secret"), - { audience: masterSession } - ); - - connector = activeConnectors.get(payload.sid); - if (!connector) { - throw new Error(); - } - - } catch (err) { - socket.send(JSON.stringify({ seq: null, type: "invalidate", data: {}})); - attachHandler(); - return; - } - - connector.continue(socket, data.last_seq); - return; - } - - if (stopAcceptingNewSessions) { - logger.verbose(`SOCKET | closing ${req.socket.remoteAddress}, server is shutting down`); - wsRejectedConnectionsCounter.inc(); - socket.close(1001, "Shutting down"); - return; - } - - // Validate token - let user; - try { - if (type === "register_slave") { - let tokenPayload = jwt.verify( - data.token, - masterConfig.get("master.auth_secret"), - { audience: "slave" } - ); - - if (tokenPayload.slave !== data.id) { - throw new Error("missmatched slave id"); - } - - } else if (type === "register_control") { - let tokenPayload = jwt.verify( - data.token, - masterConfig.get("master.auth_secret"), - { audience: "user" } - ); - - user = db.users.get(tokenPayload.user); - if (!user) { - throw new Error("invalid user"); - } - if (tokenPayload.iat < user.tokenValidAfter) { - throw new Error("invalid token"); - } - user.checkPermission("core.control.connect"); - } - - } catch (err) { - logger.verbose(`SOCKET | authentication failed for ${req.socket.remoteAddress}`); - wsRejectedConnectionsCounter.inc(); - socket.close(4003, `Authentication failed: ${err.message}`); - return; - } - - let sessionId = nextSessionId; - nextSessionId += 1; - let sessionToken = jwt.sign({ aud: masterSession, sid: sessionId }, masterConfig.get("master.auth_secret")); - let connector = new WebSocketServerConnector(socket, sessionId); - activeConnectors.set(sessionId, connector); - - let additionalReadyData = {}; - if (type === "register_slave") { - let connection = slaveConnections.get(data.id); - if (connection) { - logger.verbose(`SOCKET | disconnecting existing connection for slave ${data.id}`); - connection.connector.setTimeout(15); // Slave connection is likely stalled - await connection.disconnect(1008, "Registered from another connection"); - } - - logger.verbose(`SOCKET | registered slave ${data.id} version ${data.version}`); - slaveConnections.set(data.id, new SlaveConnection(data, connector)); - - } else if (type === "register_control") { - logger.verbose(`SOCKET | registered control from ${req.socket.remoteAddress}`); - controlConnections.push(new ControlConnection(data, connector, user)); - additionalReadyData["account"] = { - name: user.name, - }; - } - - connector.ready(sessionToken, additionalReadyData); -} - -wss.on("connection", (socket, req) => { - logger.verbose(`SOCKET | new connection from ${req.socket.remoteAddress}`); - - // Track statistics - wsConnectionsCounter.inc(); - socket.on("message", (message) => { - wsMessageCounter.labels("in").inc(); - if (!socket.clusterio_ignore_dump) { - debugEvents.emit("message", { direction: "in", content: message }); - } - }); - let originalSend = socket.send; - socket.send = (...args) => { - wsMessageCounter.labels("out").inc(); - if (typeof args[0] === "string" && !socket.clusterio_ignore_dump) { - debugEvents.emit("message", { direction: "out", content: args[0] }); - } - return originalSend.call(socket, ...args); - }; - - // Start connection handshake. - socket.send(JSON.stringify({ seq: null, type: "hello", data: { - version, - plugins: loadedPlugins, - }})); - - function attachHandler() { - pendingSockets.add(socket); - - let timeoutId = setTimeout(() => { - logger.verbose(`SOCKET | closing ${req.socket.remoteAddress} after timing out on handshake`); - wsRejectedConnectionsCounter.inc(); - socket.close(1008, "Handshake timeout"); - pendingSockets.delete(socket); - }, 30*1000); - - socket.once("message", (message) => { - clearTimeout(timeoutId); - pendingSockets.delete(socket); - handleHandshake( - message, socket, req, attachHandler - ).catch(err => { - logger.error(` -+----------------------------------------------------------------+ -| Unexpected error occured in WebSocket handshake, please report | -| it to https://github.com/clusterio/factorioClusterio/issues | -+----------------------------------------------------------------+ -${err.stack}` - ); - wsRejectedConnectionsCounter.inc(); - socket.close(1011, "Unexpected error"); - }); - }); - } - - attachHandler(); -}); - -async function loadPlugins() { - for (let pluginInfo of pluginInfos) { - if (!masterConfig.group(pluginInfo.name).get("load_plugin")) { - continue; - } - - loadedPlugins[pluginInfo.name] = pluginInfo.version; - - let pluginLoadStarted = Date.now(); - let MasterPluginClass = libPlugin.BaseMasterPlugin; - try { - if (pluginInfo.masterEntrypoint) { - MasterPluginClass = await libPluginLoader.loadMasterPluginClass(pluginInfo); - } - - let masterPlugin = new MasterPluginClass( - pluginInfo, - { - app, - config: masterConfig, - db, - getMasterUrl, - slaveConnections, - }, - { endpointHitCounter }, - logger - ); - await masterPlugin.init(); - masterPlugins.set(pluginInfo.name, masterPlugin); - - } catch (err) { - throw new libErrors.PluginError(pluginInfo.name, err); - } - - logger.info(`Loaded plugin ${pluginInfo.name} in ${Date.now() - pluginLoadStarted}ms`); - } -} - -// handle plugins on the master -async function pluginManagement() { - let startPluginLoad = Date.now(); - await loadPlugins(); - logger.info(`All plugins loaded in ${Date.now() - startPluginLoad}ms`); -} - -/** - * Calls listen on server capturing any errors that occurs - * binding to the port. Also adds handler for WebSocket - * upgrade event. - * - * @param {module:net.Server} server - Server to start the listening on. - * @param {*} args - Arguments to the .listen() call on the server. - * @returns {Promise} promise that resolves the server is listening. - */ -function listen(server, ...args) { - return new Promise((resolve, reject) => { - server.on("upgrade", (req, socket, head) => { - logger.verbose("handling WebSocket upgrade"); - - // For reasons that defy common sense, the connection event has - // to be emitted explictly when using noServer. - wss.handleUpgrade(req, socket, head, (ws) => { - wss.emit("connection", ws, req); - }); - }); - - function wrapError(err) { - reject(new libErrors.StartupError( - `Server listening failed: ${err.message}` - )); - } - - server.once("error", wrapError); - server.listen(...args, () => { - server.off("error", wrapError); - resolve(); - }); - }); -} - -function _setConfig(config) { - masterConfig = config; -} - -async function handleBootstrapCommand(args) { +async function handleBootstrapCommand(args, masterConfig) { let subCommand = args._[1]; - await loadUsers(masterConfig.get("master.database_directory"), "users.json"); + let userManager = new UserManager(masterConfig); + await userManager.load(path.join(masterConfig.get("master.database_directory"), "users.json")); if (subCommand === "create-admin") { if (!args.name) { logger.error("name cannot be blank"); @@ -1836,18 +115,18 @@ async function handleBootstrapCommand(args) { return; } - let admin = db.users.get(args.name); + let admin = userManager.users.get(args.name); if (!admin) { - admin = createUser(args.name); + admin = userManager.createUser(args.name); } - let adminRole = libUsers.ensureDefaultAdminRole(db.roles); + let adminRole = libUsers.ensureDefaultAdminRole(userManager.roles); admin.roles.add(adminRole); admin.isAdmin = true; - await saveUsers(masterConfig.get("master.database_directory"), "users.json"); + await userManager.save(path.join(masterConfig.get("master.database_directory"), "users.json")); } else if (subCommand === "generate-user-token") { - let user = db.users.get(args.name); + let user = userManager.users.get(args.name); if (!user) { logger.error(`No user named '${args.name}'`); process.exitCode = 1; @@ -1857,7 +136,7 @@ async function handleBootstrapCommand(args) { console.log(user.createToken(masterConfig.get("master.auth_secret"))); } else if (subCommand === "create-ctl-config") { - let admin = db.users.get(args.name); + let admin = userManager.users.get(args.name); if (!admin) { logger.error(`No user named '${args.name}'`); process.exitCode = 1; @@ -1866,7 +145,7 @@ async function handleBootstrapCommand(args) { let controlConfig = new libConfig.ControlConfig("control"); await controlConfig.init(); - controlConfig.set("control.master_url", getMasterUrl()); + controlConfig.set("control.master_url", Master.calculateMasterUrl(masterConfig)); controlConfig.set("control.master_token", admin.createToken(masterConfig.get("master.auth_secret"))); let content = JSON.stringify(controlConfig.serialize(), null, 4); @@ -1881,8 +160,17 @@ async function handleBootstrapCommand(args) { } async function initialize() { + let parameters = { + args: null, + shouldRun: false, + clusterLogger: null, + pluginInfos: null, + masterConfigPath: null, + masterConfig: null, + }; + // argument parsing - let args = yargs + parameters.args = yargs .scriptName("master") .usage("$0 [options]") .option("log-level", { @@ -1928,12 +216,12 @@ async function initialize() { ; // Combined log stream of the whole cluster. - clusterLogger = winston.createLogger({ + parameters.clusterLogger = winston.createLogger({ format: winston.format.json(), level: "verbose", levels, }); - clusterLogger.add(new winston.transports.File({ + parameters.clusterLogger.add(new winston.transports.File({ filename: "cluster.log", })); @@ -1943,19 +231,19 @@ async function initialize() { filename: "master.log", })); logger.add(new winston.transports.Stream({ - stream: clusterLogger, + stream: parameters.clusterLogger, })); - if (args.logLevel !== "none") { + if (parameters.args.logLevel !== "none") { logger.add(new ConsoleTransport({ - level: args.logLevel, + level: parameters.args.logLevel, format: new libLoggingUtils.TerminalFormat(), })); } - logger.info(`Loading available plugins from ${args.pluginList}`); + logger.info(`Loading available plugins from ${parameters.args.pluginList}`); let pluginList = new Map(); try { - pluginList = new Map(JSON.parse(await fs.readFile(args.pluginList))); + pluginList = new Map(JSON.parse(await fs.readFile(parameters.args.pluginList))); } catch (err) { if (err.code !== "ENOENT") { throw err; @@ -1963,83 +251,72 @@ async function initialize() { } // If the command is plugin management we don't try to load plugins - let command = args._[0]; + let command = parameters.args._[0]; if (command === "plugin") { - await libSharedCommands.handlePluginCommand(args, pluginList, args.pluginList); - return [false, args]; + await libSharedCommands.handlePluginCommand(parameters.args, pluginList, parameters.args.pluginList); + return parameters; } logger.info("Loading Plugin info"); - pluginInfos = await libPluginLoader.loadPluginInfos(pluginList); - libConfig.registerPluginConfigGroups(pluginInfos); + parameters.pluginInfos = await libPluginLoader.loadPluginInfos(pluginList); + libConfig.registerPluginConfigGroups(parameters.pluginInfos); libConfig.finalizeConfigs(); - masterConfigPath = args.config; - logger.info(`Loading config from ${masterConfigPath}`); - masterConfig = new libConfig.MasterConfig("master"); + parameters.masterConfigPath = parameters.args.config; + logger.info(`Loading config from ${parameters.masterConfigPath}`); + parameters.masterConfig = new libConfig.MasterConfig("master"); try { - await masterConfig.load(JSON.parse(await fs.readFile(masterConfigPath))); + await parameters.masterConfig.load(JSON.parse(await fs.readFile(parameters.masterConfigPath))); } catch (err) { if (err.code === "ENOENT") { logger.info("Config not found, initializing new config"); - await masterConfig.init(); + await parameters.masterConfig.init(); } else { throw err; } } - if (!masterConfig.get("master.auth_secret")) { + if (!parameters.masterConfig.get("master.auth_secret")) { logger.info("Generating new master authentication secret"); let asyncRandomBytes = util.promisify(crypto.randomBytes); let bytes = await asyncRandomBytes(256); - masterConfig.set("master.auth_secret", bytes.toString("base64")); - await fs.outputFile(masterConfigPath, JSON.stringify(masterConfig.serialize(), null, 4)); + parameters.masterConfig.set("master.auth_secret", bytes.toString("base64")); + await fs.outputFile(parameters.masterConfigPath, JSON.stringify(parameters.masterConfig.serialize(), null, 4)); } if (command === "config") { - await libSharedCommands.handleConfigCommand(args, masterConfig, masterConfigPath); - return [false, args]; + await libSharedCommands.handleConfigCommand( + parameters.args, parameters.masterConfig, parameters.masterConfigPath + ); + return parameters; } else if (command === "bootstrap") { - await handleBootstrapCommand(args); - return [false, args]; + await handleBootstrapCommand(parameters.args, parameters.masterConfig); + return parameters; } // If we get here the command was run - return [true, args]; + parameters.shouldRun = true; + return parameters; } -async function startServer(args) { - // Start webpack development server if enabled - if (args.dev || args.devPlugin) { - logger.warn("Webpack development mode enabled"); - /* eslint-disable global-require */ - const webpack = require("webpack"); - const webpackDevMiddleware = require("webpack-dev-middleware"); - const webpackConfigs = []; - if (args.dev) { - webpackConfigs.push(require("./webpack.config")({})); - } - if (args.devPlugin) { - for (let name of args.devPlugin) { - let info = pluginInfos.find(i => i.name === name); - if (!info) { - throw new libErrors.StartupError(`No plugin named ${name}`); - } - let config = require(path.posix.join(info.requirePath, "webpack.config"))({}); - config.output.publicPath = `/plugins/${name}/`; - webpackConfigs.push(config); - } - } - /* eslint-enable global-require */ +async function startup() { + // Set the process title, shows up as the title of the CMD window on windows + // and as the process name in ps/top on linux. + process.title = "clusterioMaster"; - const compiler = webpack(webpackConfigs); - devMiddleware = webpackDevMiddleware(compiler, {}); - app.use(devMiddleware); + // add better stack traces on promise rejection + process.on("unhandledRejection", err => logger.error(`Unhandled rejection:\n${err.stack}`)); + + let { args, shouldRun, clusterLogger, pluginInfos, masterConfigPath, masterConfig } = await initialize(master); + if (!shouldRun) { + return; } + master = new Master(clusterLogger, pluginInfos, masterConfigPath, masterConfig); + let secondSigint = false; process.on("SIGINT", () => { if (secondSigint) { @@ -2051,7 +328,7 @@ async function startServer(args) { secondSigint = true; logger.info("Caught interrupt signal, shutting down"); - shutdown(); + master.stop(); }); let secondSigterm = false; process.on("SIGTERM", () => { @@ -2064,110 +341,18 @@ async function startServer(args) { secondSigterm = true; logger.info("Caught termination signal, shutting down"); - shutdown(); + master.stop(); }); + process.on("SIGHUP", () => { logger.info("Terminal closed, shutting down"); - shutdown(); + master.stop(); }); - await fs.ensureDir(masterConfig.get("master.database_directory")); - - db.slaves = await loadMap(masterConfig.get("master.database_directory"), "slaves.json"); - db.instances = await loadInstances(masterConfig.get("master.database_directory"), "instances.json"); - await loadUsers(masterConfig.get("master.database_directory"), "users.json"); - - // Make sure we're actually going to listen on a port - let httpPort = masterConfig.get("master.http_port"); - let httpsPort = masterConfig.get("master.https_port"); - let bindAddress = masterConfig.get("master.bind_address") || ""; - if (!httpPort && !httpsPort) { - logger.fatal("Error: at least one of http_port and https_port must be configured"); - process.exitCode = 1; - return; - } - - let tls_cert = masterConfig.get("master.tls_certificate"); - let tls_key = masterConfig.get("master.tls_private_key"); - - if (httpsPort && (!tls_cert || !tls_key)) { - throw new libErrors.StartupError( - "tls_certificate and tls_private_key must be configure in order to use https_port" - ); - } - - // Add routes for the web interface - for (let route of routes) { - app.get(route, serveWeb(route)); - } - for (let pluginInfo of pluginInfos) { - for (let route of pluginInfo.routes || []) { - app.get(route, serveWeb(route)); - } - - let pluginPackagePath = require.resolve(path.posix.join(pluginInfo.requirePath, "package.json")); - let webPath = path.join(path.dirname(pluginPackagePath), "dist", "web"); - app.use(`/plugins/${pluginInfo.name}`, express.static(webPath)); - } - - // Load plugins - await pluginManagement(); - - // Only start listening for connections after all plugins have loaded - if (httpPort) { - httpServer = http.createServer(app); - httpServerCloser = new HttpCloser(httpServer); - await listen(httpServer, httpPort, bindAddress); - logger.info(`Listening for HTTP on port ${httpServer.address().port}`); - } - - if (httpsPort) { - let certificate, privateKey; - try { - certificate = await fs.readFile(tls_cert); - privateKey = await fs.readFile(tls_key); - - } catch (err) { - throw new libErrors.StartupError( - `Error loading ssl certificate: ${err.message}` - ); - } - - httpsServer = https.createServer({ - key: privateKey, - cert: certificate, - }, app); - httpsServerCloser = new HttpCloser(httpsServer); - await listen(httpsServer, httpsPort, bindAddress); - logger.info(`Listening for HTTPS on port ${httpsServer.address().port}`); - } -} - -async function startup() { - // Set the process title, shows up as the title of the CMD window on windows - // and as the process name in ps/top on linux. - process.title = "clusterioMaster"; - - // add better stack traces on promise rejection - process.on("unhandledRejection", err => logger.error(`Unhandled rejection:\n${err.stack}`)); - - let [shouldRun, args] = await initialize(); - if (shouldRun) { - await startServer(args); - } + await master.start(args); } module.exports = { - app, - - // For testing only - _db: db, - _setConfig, - _WebSocketServerConnector: WebSocketServerConnector, - _controlConnections: controlConnections, - _ControlConnection: ControlConnection, - _slaveConnections: slaveConnections, - _SlaveConnection: SlaveConnection, }; if (module === require.main) { @@ -2204,8 +389,6 @@ ${err.stack}` ); } - if (masterConfig) { - shutdown(); - } + process.exitCode = 1; }); } diff --git a/packages/master/package.json b/packages/master/package.json index 0ceff75e..185e5526 100644 --- a/packages/master/package.json +++ b/packages/master/package.json @@ -20,9 +20,7 @@ }, "dependencies": { "@clusterio/lib": "^2.0.0-alpha.5", - "body-parser": "^1.18.2", "compression": "^1.7.2", - "cookie-parser": "^1.4.3", "express": "^4.16.2", "express-fileupload": "1.2.0", "fs-extra": "^8.1.0", diff --git a/packages/master/src/BaseConnection.js b/packages/master/src/BaseConnection.js new file mode 100644 index 00000000..191f1bf7 --- /dev/null +++ b/packages/master/src/BaseConnection.js @@ -0,0 +1,95 @@ +"use strict"; +const libErrors = require("@clusterio/lib/errors"); +const libLink = require("@clusterio/lib/link"); +const { logger } = require("@clusterio/lib/logging"); +const libPlugin = require("@clusterio/lib/plugin"); + + +/** + * Base class for master server connections + * + * @extends module:lib/link.Link + * @alias module:master/src/BaseConnection + */ +class BaseConnection extends libLink.Link { + constructor(target, connector, master) { + /** @member {module:master/src/WsServerConnector} module:master/src/BaseConnection#connector */ + super("master", target, connector); + this._master = master; + libLink.attachAllMessages(this); + for (let masterPlugin of this._master.plugins.values()) { + libPlugin.attachPluginMessages(this, masterPlugin); + } + } + + async forwardRequestToInstance(message, request) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + } + + let slaveId = instance.config.get("instance.assigned_slave"); + if (slaveId === null) { + throw new libErrors.RequestError("Instance is not assigned to a slave"); + } + + let connection = this._master.wsServer.slaveConnections.get(slaveId); + if (!connection) { + throw new libErrors.RequestError("Slave containing instance is not connected"); + } + if (request.plugin && !connection.plugins.has(request.plugin)) { + throw new libErrors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`); + } + + return await request.send(connection, message.data); + } + + async forwardEventToInstance(message, event) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { return; } + + let slaveId = instance.config.get("instance.assigned_slave"); + if (slaveId === null) { return; } + + let connection = this._master.wsServer.slaveConnections.get(slaveId); + if (!connection || connection.closing) { return; } + if (event.plugin && !connection.plugins.has(event.plugin)) { return; } + + event.send(connection, message.data); + } + + async broadcastEventToSlaves(message, event) { + for (let slaveConnection of this._master.wsServer.slaveConnections.values()) { + // Do not broadcast back to the source + if (slaveConnection === this) { continue; } + if (slaveConnection.connector.closing) { continue; } + if (event.plugin && !slaveConnection.plugins.has(event.plugin)) { continue; } + + event.send(slaveConnection, message.data); + } + } + + async broadcastEventToInstance(message, event) { + await this.broadcastEventToSlaves(message, event); + } + + async prepareDisconnectRequestHandler(message, request) { + await libPlugin.invokeHook(this._master.plugins, "onPrepareSlaveDisconnect", this); + this.connector.setClosing(); + return await super.prepareDisconnectRequestHandler(message, request); + } + + async disconnect(code, reason) { + try { + await libLink.messages.prepareDisconnect.send(this); + } catch (err) { + if (!(err instanceof libErrors.SessionLost)) { + logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`); + } + } + + await this.connector.close(code, reason); + } +} + +module.exports = BaseConnection; diff --git a/packages/master/src/ControlConnection.js b/packages/master/src/ControlConnection.js new file mode 100644 index 00000000..b640a005 --- /dev/null +++ b/packages/master/src/ControlConnection.js @@ -0,0 +1,547 @@ +"use strict"; +const jwt = require("jsonwebtoken"); +const util = require("util"); +const winston = require("winston"); + +const libConfig = require("@clusterio/lib/config"); +const libErrors = require("@clusterio/lib/errors"); +const libLink = require("@clusterio/lib/link"); +const { logger } = require("@clusterio/lib/logging"); +const libLoggingUtils = require("@clusterio/lib/logging_utils"); +const libPlugin = require("@clusterio/lib/plugin"); +const libPrometheus = require("@clusterio/lib/prometheus"); +const libUsers = require("@clusterio/lib/users"); + +const BaseConnection = require("./BaseConnection"); + +const lastQueryLogTime = new libPrometheus.Gauge( + "clusterio_master_last_query_log_duration_seconds", + "Time in seconds the last log query took to execute." +); + +/** + * Represents the connection to a control link + * + * @extends module:master/src/BaseConnection + * @alias module:master/src/ControlConnection + */ +class ControlConnection extends BaseConnection { + constructor(registerData, connector, master, user) { + super("control", connector, master); + + this._agent = registerData.agent; + this._version = registerData.version; + + /** + * The user making this connection. + * @type {module:lib/users.User} + */ + this.user = user; + + this.logTransport = null; + this.logSubscriptions = { + all: false, + master: false, + slave_ids: [], + instance_ids: [], + }; + + this.ws_dumper = null; + this.connector.on("connect", () => { + this.connector._socket.clusterio_ignore_dump = Boolean(this.ws_dumper); + }); + this.connector.on("close", () => { + if (this.logTransport) { + this.logTransport = null; + logger.remove(this.logTransport); + } + if (this.ws_dumper) { + this._master.debugEvents.off("message", this.ws_dumper); + } + }); + + for (let event of ["connect", "drop", "resume", "close"]) { + this.connector.on(event, () => { + for (let masterPlugin of this._master.plugins.values()) { + masterPlugin.onControlConnectionEvent(this, event); + } + }); + } + } + + async getMasterConfigRequestHandler() { + return { serialized_config: this._master.config.serialize("control") }; + } + + async setMasterConfigFieldRequestHandler(message) { + this._master.config.set(message.data.field, message.data.value, "control"); + } + + async setMasterConfigPropRequestHandler(message) { + let { field, prop, value } = message.data; + this._master.config.setProp(field, prop, value, "control"); + } + + async listSlavesRequestHandler(message) { + let list = []; + for (let slave of this._master.slaves.values()) { + list.push({ + agent: slave.agent, + version: slave.version, + id: slave.id, + name: slave.name, + connected: this._master.wsServer.slaveConnections.has(slave.id), + }); + } + return { list }; + } + + generateSlaveToken(slaveId) { + return jwt.sign({ aud: "slave", slave: slaveId }, this._master.config.get("master.auth_secret")); + } + + async generateSlaveTokenRequestHandler(message) { + return { token: this.generateSlaveToken(message.data.slave_id) }; + } + + async createSlaveConfigRequestHandler(message) { + let slaveConfig = new libConfig.SlaveConfig("control"); + await slaveConfig.init(); + + slaveConfig.set("slave.master_url", this._master.getMasterUrl()); + if (message.data.id !== null) { + slaveConfig.set("slave.id", message.data.id); + } + if (message.data.name !== null) { + slaveConfig.set("slave.name", message.data.name); + } + if (message.data.generate_token) { + this.user.checkPermission("core.slave.generate_token"); + slaveConfig.set("slave.master_token", this.generateSlaveToken(slaveConfig.get("slave.id"))); + } + return { serialized_config: slaveConfig.serialize() }; + } + + async listInstancesRequestHandler(message) { + let list = []; + for (let instance of this._master.instances.values()) { + list.push({ + id: instance.config.get("instance.id"), + name: instance.config.get("instance.name"), + assigned_slave: instance.config.get("instance.assigned_slave"), + status: instance.status, + }); + } + return { list }; + } + + // XXX should probably add a hook for slave reuqests? + async createInstanceRequestHandler(message) { + let instanceConfig = new libConfig.InstanceConfig("master"); + await instanceConfig.load(message.data.serialized_config); + + let instanceId = instanceConfig.get("instance.id"); + if (this._master.instances.has(instanceId)) { + throw new libErrors.RequestError(`Instance with ID ${instanceId} already exists`); + } + + // Add common settings for the Factorio server + let settings = { + "name": `${this._master.config.get("master.name")} - ${instanceConfig.get("instance.name")}`, + "description": `Clusterio instance for ${this._master.config.get("master.name")}`, + "tags": ["clusterio"], + "max_players": 0, + "visibility": { "public": true, "lan": true }, + "username": "", + "token": "", + "game_password": "", + "require_user_verification": true, + "max_upload_in_kilobytes_per_second": 0, + "max_upload_slots": 5, + "ignore_player_limit_for_returning_players": false, + "allow_commands": "admins-only", + "autosave_interval": 10, + "autosave_slots": 5, + "afk_autokick_interval": 0, + "auto_pause": false, + "only_admins_can_pause_the_game": true, + "autosave_only_on_server": true, + + ...instanceConfig.get("factorio.settings"), + }; + instanceConfig.set("factorio.settings", settings); + + let instance = { config: instanceConfig, status: "unassigned" }; + this._master.instances.set(instanceId, instance); + await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, null); + this._master.addInstancePluginHooks(instance); + } + + async deleteInstanceRequestHandler(message, request) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + } + + if (instance.config.get("instance.assigned_slave") !== null) { + await this.forwardRequestToInstance(message, request); + } + this._master.instances.delete(message.data.instance_id); + + let prev = instance.status; + instance.status = "deleted"; + await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev); + } + + async getInstanceConfigRequestHandler(message) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + } + + return { + serialized_config: instance.config.serialize("control"), + }; + } + + async updateInstanceConfig(instance) { + let slaveId = instance.config.get("instance.assigned_slave"); + if (slaveId) { + let connection = this._master.wsServer.slaveConnections.get(slaveId); + if (connection) { + await libLink.messages.assignInstance.send(connection, { + instance_id: instance.config.get("instance.id"), + serialized_config: instance.config.serialize("slave"), + }); + } + } + } + + async setInstanceConfigFieldRequestHandler(message) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + } + + if (message.data.field === "instance.assigned_slave") { + throw new libErrors.RequestError("instance.assigned_slave must be set through the assign-slave interface"); + } + + if (message.data.field === "instance.id") { + // XXX is this worth implementing? It's race condition galore. + throw new libErrors.RequestError("Setting instance.id is not supported"); + } + + instance.config.set(message.data.field, message.data.value, "control"); + await this.updateInstanceConfig(instance); + } + + async setInstanceConfigPropRequestHandler(message) { + let instance = this._master.instances.get(message.data.instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + } + + let { field, prop, value } = message.data; + instance.config.setProp(field, prop, value, "control"); + await this.updateInstanceConfig(instance); + } + + async assignInstanceCommandRequestHandler(message, request) { + let { slave_id, instance_id } = message.data; + let instance = this._master.instances.get(instance_id); + if (!instance) { + throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`); + } + + // Check if target slave is connected + let newSlaveConnection; + if (slave_id !== null) { + newSlaveConnection = this._master.wsServer.slaveConnections.get(slave_id); + if (!newSlaveConnection) { + // The case of the slave not getting the assign instance message + // still have to be handled, so it's not a requirement that the + // target slave be connected to the master while doing the + // assignment, but it is IMHO a better user experience if this + // is the case. + throw new libErrors.RequestError("Target slave is not connected to the master server"); + } + } + + // Unassign from currently assigned slave if it is connected. + let currentAssignedSlave = instance.config.get("instance.assigned_slave"); + if (currentAssignedSlave !== null && slave_id !== currentAssignedSlave) { + let oldSlaveConnection = this._master.wsServer.slaveConnections.get(currentAssignedSlave); + if (oldSlaveConnection && !oldSlaveConnection.connector.closing) { + await libLink.messages.unassignInstance.send(oldSlaveConnection, { instance_id }); + } + } + + // Assign to target + instance.config.set("instance.assigned_slave", slave_id); + if (slave_id !== null) { + await libLink.messages.assignInstance.send(newSlaveConnection, { + instance_id, + serialized_config: instance.config.serialize("slave"), + }); + } + } + + async setLogSubscriptionsRequestHandler(message) { + this.logSubscriptions = message.data; + this.updateLogSubscriptions(); + } + + static logFilter({ all, master, slave_ids, instance_ids, max_level }) { + return info => { + // Note: reversed to filter out undefined levels + if (max_level && !(levels[info.level] <= levels[max_level])) { + return false; + } + + if (all) { + return true; + } + if (master && info.slave_id === undefined) { + return true; + } + if (info.slave_id && slave_ids.includes(info.slave_id)) { + return true; + } + if (info.instance_id && instance_ids.includes(info.instance_id)) { + return true; + } + return false; + }; + } + + updateLogSubscriptions() { + let { all, master, slave_ids, instance_ids } = this.logSubscriptions; + if (all || master || slave_ids.length || instance_ids.length) { + if (!this.logTransport) { + this.logTransport = new libLoggingUtils.LinkTransport({ link: this }); + this._master.clusterLogger.add(this.logTransport); + } + this.logTransport.filter = this.constructor.logFilter(this.logSubscriptions); + + } else if (this.logTransport) { + this._master.clusterLogger.remove(this.logTransport); + this.logTransport = null; + } + } + + async queryLogRequestHandler(message) { + let transport = new winston.transports.File({ filename: "cluster.log" }); + let query = util.promisify((options, callback) => transport.query(options, callback)); + + // Available options and defaults inferred from reading the source + // rows: number = limit || 10 + // limit: number // alias of rows. + // start: number = 0 + // until: Date = now + // from: Date = until - 24h + // order: "asc"|"desc" = "desc" + // fields: Array + // level: string + // This interface is junk: + // - The level option filters by exact match + // - No way to add your own filter + // - No way to stream results + // - The log file is read starting from the beginning + let setDuration = lastQueryLogTime.startTimer(); + let log = await query({ + order: "asc", + limit: Infinity, + from: new Date(0), + }); + + log = log.filter(this.constructor.logFilter(message.data)); + setDuration(); + return { log }; + } + + async listPermissionsRequestHandler(message) { + let list = []; + for (let permission of libUsers.permissions.values()) { + list.push({ + name: permission.name, + title: permission.title, + description: permission.description, + }); + } + return { list }; + } + + async listRolesRequestHandler(message) { + let list = []; + for (let role of this._master.userManager.roles.values()) { + list.push({ + id: role.id, + name: role.name, + description: role.description, + permissions: [...role.permissions], + }); + } + return { list }; + } + + async createRoleRequestHandler(message) { + let lastId = Math.max.apply(null, [...this._master.userManager.roles.keys()]); + + // Start at 5 to leave space for future default roles + let id = Math.max(5, lastId+1); + this._master.userManager.roles.set(id, new libUsers.Role({ id, ...message.data })); + return { id }; + } + + async updateRoleRequestHandler(message) { + let { id, name, description, permissions } = message.data; + let role = this._master.userManager.roles.get(id); + if (!role) { + throw new libErrors.RequestError(`Role with ID ${id} does not exist`); + } + + role.name = name; + role.description = description; + role.permissions = new Set(permissions); + } + + async grantDefaultRolePermissionsRequestHandler(message) { + let role = this._master.userManager.roles.get(message.data.id); + if (!role) { + throw new libErrors.RequestError(`Role with ID ${message.data.id} does not exist`); + } + + role.grantDefaultPermissions(); + } + + async deleteRoleRequestHandler(message) { + let id = message.data.id; + let role = this._master.userManager.roles.get(id); + if (!role) { + throw new libErrors.RequestError(`Role with ID ${id} does not exist`); + } + + this._master.userManager.roles.delete(id); + for (let user of this._master.userManager.users.values()) { + user.roles.delete(role); + } + } + + async listUsersRequestHandler(message) { + let list = []; + for (let user of this._master.userManager.users.values()) { + list.push({ + name: user.name, + roles: [...user.roles].map(role => role.id), + is_admin: user.isAdmin, + is_banned: user.isBanned, + is_whitelisted: user.isWhitelisted, + instances: [...user.instances], + }); + } + return { list }; + } + + async createUserRequestHandler(message) { + this._master.userManager.createUser(message.data.name); + } + + async updateUserRolesRequestHandler(message) { + let user = this._master.userManager.users.get(message.data.name); + if (!user) { + throw new libErrors.RequestError(`User '${message.data.name}' does not exist`); + } + + let resolvedRoles = new Set(); + for (let roleId of message.data.roles) { + let role = this._master.userManager.roles.get(roleId); + if (!role) { + throw new libErrors.RequestError(`Role with ID ${roleId} does not exist`); + } + + resolvedRoles.add(role); + } + + user.roles = resolvedRoles; + } + + async setUserAdminRequestHandler(message) { + let { name, create, admin } = message.data; + let user = this._master.userManager.users.get(name); + if (!user) { + if (create) { + this.user.checkPermission("core.user.create"); + user = this._master.userManager.createUser(name); + } else { + throw new libErrors.RequestError(`User '${name}' does not exist`); + } + } + + user.isAdmin = admin; + this.broadcastEventToSlaves({ data: { name, admin }}, libLink.messages.adminlistUpdate); + } + + async setUserBannedRequestHandler(message) { + let { name, create, banned, reason } = message.data; + let user = this._master.userManager.users.get(name); + if (!user) { + if (create) { + this.user.checkPermission("core.user.create"); + user = this._master.userManager.createUser(name); + } else { + throw new libErrors.RequestError(`User '${name}' does not exist`); + } + } + + user.isBanned = banned; + user.banReason = reason; + this.broadcastEventToSlaves({ data: { name, banned, reason }}, libLink.messages.banlistUpdate); + } + + async setUserWhitelistedRequestHandler(message) { + let { name, create, whitelisted } = message.data; + let user = this._master.userManager.users.get(name); + if (!user) { + if (create) { + this.user.checkPermission("core.user.create"); + user = this._master.userManager.createUser(name); + } else { + throw new libErrors.RequestError(`User '${name}' does not exist`); + } + } + + user.isWhitelisted = whitelisted; + this.broadcastEventToSlaves({ data: { name, whitelisted }}, libLink.messages.whitelistUpdate); + } + + async deleteUserRequestHandler(message) { + let user = this._master.userManager.users.get(message.data.name); + if (!user) { + throw new libErrors.RequestError(`User '${message.data.name}' does not exist`); + } + + if (user.is_admin) { + this.broadcastEventToSlaves({ data: { name, admin: false }}, libLink.messages.adminlistUpdate); + } + if (user.is_whitelisted) { + this.broadcastEventToSlaves({ data: { name, whitelisted: false }}, libLink.messages.whitelistUpdate); + } + if (user.is_banned) { + this.broadcastEventToSlaves({ data: { name, banned: false, reason: "" }}, libLink.messages.banlistUpdate); + } + this._master.userManager.users.delete(message.data.name); + } + + async debugDumpWsRequestHandler(message) { + this.ws_dumper = data => { + if (this.connector.connected) { + libLink.messages.debugWsMessage.send(this, data); + } + }; + this.connector._socket.clusterio_ignore_dump = true; + this._master.debugEvents.on("message", this.ws_dumper); + } +} + +module.exports = ControlConnection; diff --git a/packages/master/src/Master.js b/packages/master/src/Master.js new file mode 100644 index 00000000..b6741603 --- /dev/null +++ b/packages/master/src/Master.js @@ -0,0 +1,471 @@ +"use strict"; +const compression = require("compression"); +const events = require("events"); +const express = require("express"); +const expressFileupload = require("express-fileupload"); +const fs = require("fs-extra"); +const http = require("http"); +const https = require("https"); +const path = require("path"); + + +const libConfig = require("@clusterio/lib/config"); +const libErrors = require("@clusterio/lib/errors"); +const libPlugin = require("@clusterio/lib/plugin"); +const libPluginLoader = require("@clusterio/lib/plugin_loader"); +const libPrometheus = require("@clusterio/lib/prometheus"); +const { logger } = require("@clusterio/lib/logging"); + +const HttpCloser = require("./HttpCloser"); +const metrics = require("./metrics"); +const routes = require("./routes"); +const UserManager = require("./UserManager"); +const WsServer = require("./WsServer"); + + +const hitCounter = new libPrometheus.Counter( + "clusterio_master_http_hits_total", + "How many HTTP requests in total have been received" +); + +/** + * Manages all master related operations + * @alias module:master/src/Master + */ +class Master { + constructor(clusterLogger, pluginInfos, configPath, config) { + this.clusterLogger = clusterLogger; + /** + * Mapping of plugin name to info objects for known plugins + * @type {Map} + */ + this.pluginInfos = pluginInfos; + this.configPath = configPath; + /** + * Master config. + * @type {module:lib/config.MasterConfig} + */ + this.config = config; + + this.app = express(); + this.app.locals.master = this; + + /** + * Mapping of slave id to slave info + * @type {Map} + */ + this.slaves = null; + + /** + * Mapping of instance id to instance info + * @type {Map} + */ + this.instances = null; + + /** + * User and roles manager for the cluster + * @type {module:master/src/UserManager} + */ + this.userManager = null; + this.httpServer = null; + this.httpServerCloser = null; + this.httpsServer = null; + this.httpsServerCloser = null; + + /** + * Mapping of plugin name to loaded plugin + * @type {Map} + */ + this.plugins = new Map(); + + /** + * WebSocket server + * @type {module:master/src/WsServer} + */ + this.wsServer = new WsServer(this); + + this.debugEvents = new events.EventEmitter(); + this._events = new events.EventEmitter(); + // Possible states are new, starting, running, stopping, stopped + this._state = "new"; + this._shouldStop = false; + + this.devMiddleware = null; + } + + async start(args) { + if (!this._state === "new") { + throw new Error(`Cannot start in state ${this._state}`); + } + + this._state = "starting"; + try { + await this._startInternal(args); + + } catch (err) { + await this._stopInternal(); + this._state = "stopped"; + this._events.emit("stop"); + throw err; + } + + if (this._shouldStop) { + this.stop(); + } + } + + async _startInternal(args) { + // Start webpack development server if enabled + if (args.dev || args.devPlugin) { + logger.warn("Webpack development mode enabled"); + /* eslint-disable global-require */ + const webpack = require("webpack"); + const webpackDevMiddleware = require("webpack-dev-middleware"); + const webpackConfigs = []; + if (args.dev) { + webpackConfigs.push(require("../webpack.config")({})); + } + if (args.devPlugin) { + for (let name of args.devPlugin) { + let info = this.pluginInfos.find(i => i.name === name); + if (!info) { + throw new libErrors.StartupError(`No plugin named ${name}`); + } + let config = require(path.posix.join(info.requirePath, "webpack.config"))({}); + config.output.publicPath = `/plugins/${name}/`; + webpackConfigs.push(config); + } + } + /* eslint-enable global-require */ + + const compiler = webpack(webpackConfigs); + this.devMiddleware = webpackDevMiddleware(compiler, {}); + app.use(this.devMiddleware); + } + + let databaseDirectory = this.config.get("master.database_directory"); + await fs.ensureDir(databaseDirectory); + + this.slaves = await Master.loadSlaves(path.join(databaseDirectory, "slaves.json")); + this.instances = await Master.loadInstances(path.join(databaseDirectory, "instances.json")); + this.userManager = new UserManager(this.config); + await this.userManager.load(path.join(databaseDirectory, "users.json")); + + for (let instance of this.instances.values()) { + this.addInstancePluginHooks(instance); + } + + // Make sure we're actually going to listen on a port + let httpPort = this.config.get("master.http_port"); + let httpsPort = this.config.get("master.https_port"); + let bindAddress = this.config.get("master.bind_address") || ""; + if (!httpPort && !httpsPort) { + logger.fatal("Error: at least one of http_port and https_port must be configured"); + process.exitCode = 1; + return; + } + + let tls_cert = this.config.get("master.tls_certificate"); + let tls_key = this.config.get("master.tls_private_key"); + + if (httpsPort && (!tls_cert || !tls_key)) { + throw new libErrors.StartupError( + "tls_certificate and tls_private_key must be configure in order to use https_port" + ); + } + + Master.addAppRoutes(this.app, this.pluginInfos); + + // Load plugins + await this.loadPlugins(); + + this.wsServer = new WsServer(this); + + // Only start listening for connections after all plugins have loaded + if (httpPort) { + this.httpServer = http.createServer(this.app); + this.httpServerCloser = new HttpCloser(this.httpServer); + await this.listen(this.httpServer, httpPort, bindAddress); + logger.info(`Listening for HTTP on port ${this.httpServer.address().port}`); + } + + if (httpsPort) { + let certificate, privateKey; + try { + certificate = await fs.readFile(tls_cert); + privateKey = await fs.readFile(tls_key); + + } catch (err) { + throw new libErrors.StartupError( + `Error loading ssl certificate: ${err.message}` + ); + } + + this.httpsServer = https.createServer({ + key: privateKey, + cert: certificate, + }, this.app); + this.httpsServerCloser = new HttpCloser(this.httpsServer); + await this.listen(this.httpsServer, httpsPort, bindAddress); + logger.info(`Listening for HTTPS on port ${this.httpsServer.address().port}`); + } + + logger.info("Started master"); + this._state = "running"; + } + + /** + * Stops the master server. + * + * Save data and bring down the active connections. This is the reverse + * of start and is valid at any point in time. + */ + async stop() { + if (this._state === "starting") { + this._shouldStop = true; + await events.once(this._events, "stop"); + return; + + } else if (this._state === "running") { + await this._stopInternal(); + } + + this._state = "stopped"; + this._events.emit("stop"); + } + + async _stopInternal() { + // This function should never throw. + this._state = "stopping"; + logger.info("Stopping master"); + + logger.info("Saving config"); + await fs.outputFile(this.configPath, JSON.stringify(this.config.serialize(), null, 4)); + + if (this.devMiddleware) { + await new Promise((resolve, reject) => { devMiddleware.close(resolve); }); + } + + let databaseDirectory = this.config.get("master.database_directory"); + if (this.slaves) { + await Master.saveSlaves(path.join(databaseDirectory, "slaves.json"), this.slaves); + } + + if (this.instances) { + await Master.saveInstances(path.join(databaseDirectory, "instances.json"), this.instances); + } + + if (this.userManager) { + await this.userManager.save(path.join(databaseDirectory, "users.json")); + } + + await libPlugin.invokeHook(this.plugins, "onShutdown"); + + if (this.wsServer) { + await this.wsServer.stop(); + } + + let stopTasks = []; + logger.info("Stopping HTTP(S) server"); + if (this.httpServer && this.httpServer.listening) { stopTasks.push(this.httpServerCloser.close()); } + if (this.httpsServer && this.httpsServer.listening) { stopTasks.push(this.httpsServerCloser.close()); } + await Promise.all(stopTasks); + + logger.info("Goodbye"); + } + + static async loadSlaves(filePath) { + let serialized; + try { + serialized = JSON.parse(await fs.readFile(filePath)); + + } catch (err) { + if (err.code !== "ENOENT") { + throw err; + } + + return new Map(); + } + + // TODO: Remove after release. + if (serialized.length && !(serialized[0] instanceof Array)) { + return new Map(); // Discard old format. + } + + return new Map(serialized); + } + + static async saveSlaves(filePath, slaves) { + await fs.outputFile(filePath, JSON.stringify([...slaves.entries()], null, 4)); + } + + static async loadInstances(filePath) { + logger.info(`Loading ${filePath}`); + + let instances = new Map(); + try { + let serialized = JSON.parse(await fs.readFile(filePath)); + for (let serializedConfig of serialized) { + let instanceConfig = new libConfig.InstanceConfig("master"); + await instanceConfig.load(serializedConfig); + let status = instanceConfig.get("instance.assigned_slave") === null ? "unassigned" : "unknown"; + let instance = { config: instanceConfig, status }; + instances.set(instanceConfig.get("instance.id"), instance); + } + + } catch (err) { + if (err.code !== "ENOENT") { + throw err; + } + } + + return instances; + } + + static async saveInstances(filePath, instances) { + let serialized = []; + for (let instance of instances.values()) { + serialized.push(instance.config.serialize()); + } + + await fs.outputFile(filePath, JSON.stringify(serialized, null, 4)); + } + + static addAppRoutes(app, pluginInfos) { + app.use((req, res, next) => { + hitCounter.inc(); + next(); + }); + app.use(expressFileupload()); + app.use(compression()); + + // Set folder to serve static content from (the website) + app.use(express.static(path.join(__dirname, "..", "dist", "web"))); + app.use(express.static("static")); // Used for data export files + + // Add API routes + routes.addRouteHandlers(app); + + // Add routes for the web interface + for (let route of routes.webRoutes) { + app.get(route, Master.serveWeb(route)); + } + for (let pluginInfo of pluginInfos) { + for (let route of pluginInfo.routes || []) { + app.get(route, Master.serveWeb(route)); + } + + let pluginPackagePath = require.resolve(path.posix.join(pluginInfo.requirePath, "package.json")); + let webPath = path.join(path.dirname(pluginPackagePath), "dist", "web"); + app.use(`/plugins/${pluginInfo.name}`, express.static(webPath)); + } + } + + addInstancePluginHooks(instance) { + instance.config.on("fieldChanged", (group, field, prev) => { + libPlugin.invokeHook(this.plugins, "onInstanceConfigFieldChanged", instance, group, field, prev); + }); + } + + async loadPlugins() { + for (let pluginInfo of this.pluginInfos) { + if (!this.config.group(pluginInfo.name).get("load_plugin")) { + continue; + } + + let MasterPluginClass = libPlugin.BaseMasterPlugin; + try { + if (pluginInfo.masterEntrypoint) { + MasterPluginClass = await libPluginLoader.loadMasterPluginClass(pluginInfo); + } + + let masterPlugin = new MasterPluginClass(pluginInfo, this, metrics, logger); + await masterPlugin.init(); + this.plugins.set(pluginInfo.name, masterPlugin); + + } catch (err) { + throw new libErrors.PluginError(pluginInfo.name, err); + } + + logger.info(`Loaded plugin ${pluginInfo.name}`); + } + } + + /** + * Calls listen on server capturing any errors that occurs + * binding to the port. Also adds handler for WebSocket + * upgrade event. + * + * @param {module:net.Server} server - Server to start the listening on. + * @param {*} args - Arguments to the .listen() call on the server. + * @returns {Promise} promise that resolves the server is listening. + */ + listen(server, ...args) { + return new Promise((resolve, reject) => { + server.on("upgrade", (req, socket, head) => { + logger.verbose("handling WebSocket upgrade"); + this.wsServer.handleUpgrade(req, socket, head); + }); + + function wrapError(err) { + reject(new libErrors.StartupError( + `Server listening failed: ${err.message}` + )); + } + + server.once("error", wrapError); + server.listen(...args, () => { + server.off("error", wrapError); + resolve(); + }); + }); + } + + + /** + * Returns the URL needed to connect to the master server. + * + * @returns {string} master URL. + */ + getMasterUrl() { + return Master.calculateMasterUrl(this.config); + } + + static calculateMasterUrl(config) { + let url = config.get("master.external_address"); + if (!url) { + if (config.get("master.https_port")) { + url = `https://localhost:${config.get("master.https_port")}/`; + } else { + url = `http://localhost:${config.get("master.http_port")}/`; + } + } + return url; + } + + /** + * Servers the web interface with the root path set apropriately + * + * @param {string} route - route the interface is served under. + * @returns {function} Experess.js route handler. + */ + static serveWeb(route) { + // The depth is is the number of slashes in the route minus one, but due + // to lenient matching on trailing slashes in the route we need to + // compensate if the request path contains a slash but not the route, + // and vice versa. + let routeDepth = (route.match(/\//g) || []).length - 1 - (route.slice(-1) === "/"); + return function(req, res, next) { + let depth = routeDepth + (req.path.slice(-1) === "/"); + let webRoot = "../".repeat(depth) || "./"; + fs.readFile(path.join(__dirname, "..", "web", "index.html"), "utf8").then((content) => { + res.type("text/html"); + res.send(content.replace(/__WEB_ROOT__/g, webRoot)); + }).catch(err => { + next(err); + }); + }; + } +} + +module.exports = Master; diff --git a/packages/master/src/SlaveConnection.js b/packages/master/src/SlaveConnection.js new file mode 100644 index 00000000..9bbdfd4e --- /dev/null +++ b/packages/master/src/SlaveConnection.js @@ -0,0 +1,164 @@ +"use strict"; +const libConfig = require("@clusterio/lib/config"); +const libLink = require("@clusterio/lib/link"); +const { logger } = require("@clusterio/lib/logging"); +const libPlugin = require("@clusterio/lib/plugin"); + +const BaseConnection = require("./BaseConnection"); + + +/** + * Represents the connection to a slave + * + * @extends module:master/src/BaseConnection + * @alias module:master/src/SlaveConnection + */ +class SlaveConnection extends BaseConnection { + constructor(registerData, connector, master) { + super("slave", connector, master); + + this._agent = registerData.agent; + this._id = registerData.id; + this._name = registerData.name; + this._version = registerData.version; + this.plugins = new Map(Object.entries(registerData.plugins)); + + this._master.slaves.set(this._id, { + agent: this._agent, + id: this._id, + name: this._name, + version: this._version, + plugins: registerData.plugins, + }); + + for (let event of ["connect", "drop", "resume", "close"]) { + // eslint-disable-next-line no-loop-func + this.connector.on(event, () => { + for (let plugin of this._master.plugins.values()) { + plugin.onSlaveConnectionEvent(this, event); + } + }); + } + } + + /** + * ID of the slave this connection is connected to + * + * @returns {number} slave ID. + */ + get id() { + return this._id; + } + + async instanceStatusChangedEventHandler(message, event) { + let instance = this._master.instances.get(message.data.instance_id); + + // It's possible to get updates from an instance that does not exist + // or is not assigned to the slave it originated from if it was + // reassigned or deleted while the connection to the slave it was + // originally on was down at the time. + if (!instance || instance.config.get("instance.assigned_slave") !== this._id) { + logger.warn(`Got bogus update for instance id ${message.data.instance_id}`); + return; + } + + let prev = instance.status; + instance.status = message.data.status; + logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`); + await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", instance, prev); + } + + async updateInstancesRequestHandler(message) { + // Push updated instance configs + for (let instance of this._master.instances.values()) { + if (instance.config.get("instance.assigned_slave") === this._id) { + await libLink.messages.assignInstance.send(this, { + instance_id: instance.config.get("instance.id"), + serialized_config: instance.config.serialize("slave"), + }); + } + } + + // Assign instances the slave has but master does not + for (let instanceData of message.data.instances) { + let instanceConfig = new libConfig.InstanceConfig("master"); + await instanceConfig.load(instanceData.serialized_config, "slave"); + + let masterInstance = this._master.instances.get(instanceConfig.get("instance.id")); + if (masterInstance) { + // Check if this instance is assigned somewhere else. + if (masterInstance.config.get("instance.assigned_slave") !== this._id) { + await libLink.messages.unassignInstance.send(this, { + instance_id: masterInstance.config.get("instance.id"), + }); + continue; + } + + // Already have this instance, update state instead + if (masterInstance.status !== instanceData.status) { + let prev = masterInstance.status; + masterInstance.status = instanceData.status; + logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`); + await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", masterInstance, prev); + } + continue; + } + + instanceConfig.set("instance.assigned_slave", this._id); + let newInstance = { config: instanceConfig, status: instanceData.status }; + this._master.instances.set(instanceConfig.get("instance.id"), newInstance); + this._master.addInstancePluginHooks(newInstance); + await libLink.messages.assignInstance.send(this, { + instance_id: instanceConfig.get("instance.id"), + serialized_config: instanceConfig.serialize("slave"), + }); + await libPlugin.invokeHook(this._master.plugins, "onInstanceStatusChanged", newInstance, null); + } + + // Push lists to make sure they are in sync. + let adminlist = []; + let banlist = []; + let whitelist = []; + + for (let user of this._master.userManager.users.values()) { + if (user.isAdmin) { + adminlist.push(user.name); + } + if (user.isBanned) { + banlist.push([user.name, user.banReason]); + } + if (user.isWhitelisted) { + whitelist.push(user.name); + } + } + + libLink.messages.syncUserLists.send(this, { adminlist, banlist, whitelist }); + } + + async logMessageEventHandler(message) { + this._master.clusterLogger.log({ + ...message.data.info, + slave_id: this._id, + slave_name: this._name, + }); + } + + async playerEventEventHandler(message) { + let { instance_id, name, type } = message.data; + let user = this._master.userManager.users.get(name); + if (!user) { + user = this._master.userManager.createUser(name); + } + + if (type === "join") { + user.notifyJoin(instance_id); + } else if (type === "leave") { + user.notifyLeave(instance_id); + } + + let instance = this._master.instances.get(instance_id); + await libPlugin.invokeHook(this._master.plugins, "onPlayerEvent", instance, message.data); + } +} + +module.exports = SlaveConnection; diff --git a/packages/master/src/UserManager.js b/packages/master/src/UserManager.js new file mode 100644 index 00000000..e0fa8288 --- /dev/null +++ b/packages/master/src/UserManager.js @@ -0,0 +1,88 @@ +"use strict"; +const fs = require("fs-extra"); + +const libUsers = require("@clusterio/lib/users"); + +/** + * Manages users and roles + * @alias module:master/src/UserManager + */ +class UserManager { + constructor(config) { + this._config = config; + this.roles = null; + this.users = null; + } + + async load(filePath) { + let loadedRoles = new Map(); + let loadedUsers = new Map(); + try { + let content = JSON.parse(await fs.readFile(filePath)); + for (let serializedRole of content.roles) { + let role = new libUsers.Role(serializedRole); + loadedRoles.set(role.id, role); + } + + for (let serializedUser of content.users) { + let user = new libUsers.User(serializedUser, loadedRoles); + loadedUsers.set(user.name, user); + } + + } catch (err) { + if (err.code !== "ENOENT") { + throw err; + } + + // Create default roles if loading failed + libUsers.ensureDefaultAdminRole(loadedRoles); + libUsers.ensureDefaultPlayerRole(loadedRoles); + } + + this.roles = loadedRoles; + this.users = loadedUsers; + } + + async save(filePath) { + if (!this.roles || !this.users) { + return; + } + + let serializedRoles = []; + for (let role of this.roles.values()) { + serializedRoles.push(role.serialize()); + } + + let serializedUsers = []; + for (let user of this.users.values()) { + serializedUsers.push(user.serialize()); + } + + let serialized = { + users: serializedUsers, + roles: serializedRoles, + }; + await fs.outputFile(filePath, JSON.stringify(serialized, null, 4)); + } + + /** + * Create a new user + * + * Creates a new user and add it to the user database. + * + * @param {string} name - Name of the user to create. + * @returns {module:lib/users.User} newly created user. + */ + createUser(name) { + if (this.users.has(name)) { + throw new Error(`User '${name}' already exists`); + } + + let defaultRoleId = this._config.get("master.default_role_id"); + let user = new libUsers.User({ name, roles: [defaultRoleId] }, this.roles); + this.users.set(name, user); + return user; + } +} + +module.exports = UserManager; diff --git a/packages/master/src/WsServer.js b/packages/master/src/WsServer.js new file mode 100644 index 00000000..3c77824f --- /dev/null +++ b/packages/master/src/WsServer.js @@ -0,0 +1,302 @@ +"use strict"; +const jwt = require("jsonwebtoken"); +const WebSocket = require("ws"); + +const { logger } = require("@clusterio/lib/logging"); +const libPrometheus = require("@clusterio/lib/prometheus"); +const libSchema = require("@clusterio/lib/schema"); + +const ControlConnection = require("./ControlConnection"); +const packageVersion = require("../package").version; +const SlaveConnection = require("./SlaveConnection"); +const WsServerConnector = require("./WsServerConnector"); + + +const wsMessageCounter = new libPrometheus.Counter( + "clusterio_master_websocket_message_total", + "How many messages have been received over WebSocket on the master server", + { labels: ["direction"] } +); + +const wsConnectionsCounter = new libPrometheus.Counter( + "clusterio_master_websocket_connections_total", + "How many WebSocket connections have been initiated on the master server" +); + +const wsRejectedConnectionsCounter = new libPrometheus.Counter( + "clusterio_master_websocket_rejected_connections_total", + "How many WebSocket connections have been rejected during the handshake on the master server" +); + +/** + * WebSocket server + * @alias module:master/src/WsServer + */ +class WsServer { + constructor(master) { + this.master = master; + + this.stopAcceptingNewSessions = false; + + /** @type {Array} */ + this.controlConnections = []; + /** @type {Map} */ + this.slaveConnections = new Map(); + /** @type {Map} */ + this.activeConnectors = new Map(); + /** @type {Set} */ + this.pendingSockets = new Set(); + + // Unique string for the session token audience + this.sessionAud = `session-${Date.now()}`; + this.nextSessionId = 1; + + /** @type {module:ws.Server} */ + this.wss = new WebSocket.Server({ + noServer: true, + path: "/api/socket", + }); + this.wss.on("connection", (socket, req) => this.handleConnection(socket, req)); + } + + async stop() { + this.stopAcceptingNewSessions = true; + + let disconnectTasks = []; + for (let controlConnection of this.controlConnections) { + controlConnection.connector.setTimeout(this.master.config.get("master.connector_shutdown_timeout")); + disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit")); + } + + for (let slaveConnection of this.slaveConnections.values()) { + slaveConnection.connector.setTimeout(this.master.config.get("master.connector_shutdown_timeout")); + disconnectTasks.push(slaveConnection.disconnect(1001, "Server Quit")); + } + + logger.info(`WsServer | Waiting for ${disconnectTasks.length} connectors to close`); + for (let task of disconnectTasks) { + try { + await task; + } catch (err) { + if (!(err instanceof libErrors.SessionLost)) { + logger.warn(`Unexpected error disconnecting connector:\n${err.stack}`); + } + } + } + + for (let socket of this.pendingSockets) { + socket.close(1001, "Server Quit"); + } + } + + handleUpgrade(req, socket, head) { + // For reasons that defy common sense, the connection event has + // to be emitted explictly when using noServer. + this.wss.handleUpgrade(req, socket, head, (ws) => { + this.wss.emit("connection", ws, req); + }); + } + + handleConnection(socket, req) { + logger.verbose(`WsServer | new connection from ${req.socket.remoteAddress}`); + + // Track statistics + wsConnectionsCounter.inc(); + socket.on("message", (message) => { + wsMessageCounter.labels("in").inc(); + if (!socket.clusterio_ignore_dump) { + this.master.debugEvents.emit("message", { direction: "in", content: message }); + } + }); + let originalSend = socket.send; + socket.send = (...args) => { + wsMessageCounter.labels("out").inc(); + if (typeof args[0] === "string" && !socket.clusterio_ignore_dump) { + this.master.debugEvents.emit("message", { direction: "out", content: args[0] }); + } + return originalSend.call(socket, ...args); + }; + + // Start connection handshake. + let loadedPlugins = {}; + for (let [name, plugin] of this.master.plugins) { + loadedPlugins[name] = plugin.info.version; + } + + socket.send(JSON.stringify({ seq: null, type: "hello", data: { + version: packageVersion, + plugins: loadedPlugins, + }})); + + this.attachHandler(socket, req); + } + + attachHandler(socket, req) { + this.pendingSockets.add(socket); + + let timeoutId = setTimeout(() => { + logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after timing out on handshake`); + wsRejectedConnectionsCounter.inc(); + socket.close(1008, "Handshake timeout"); + this.pendingSockets.delete(socket); + }, 30*1000); + + socket.once("message", (message) => { + clearTimeout(timeoutId); + this.pendingSockets.delete(socket); + this.handleHandshake( + message, socket, req + ).catch(err => { + logger.error(` ++----------------------------------------------------------------+ +| Unexpected error occured in WebSocket handshake, please report | +| it to https://github.com/clusterio/factorioClusterio/issues | ++----------------------------------------------------------------+ +${err.stack}` + ); + wsRejectedConnectionsCounter.inc(); + socket.close(1011, "Unexpected error"); + }); + }); + } + + async handleHandshake(message, socket, req) { + try { + message = JSON.parse(message); + } catch (err) { + logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid JSON`); + wsRejectedConnectionsCounter.inc(); + socket.close(1002, "Invalid JSON"); + return; + } + + if (!libSchema.clientHandshake(message)) { + logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid handshake`); + wsRejectedConnectionsCounter.inc(); + socket.close(1002, "Bad handshake"); + return; + } + + let { seq, type, data } = message; + + if (type === "resume") { + let connector; + try { + let payload = jwt.verify( + data.session_token, + this.master.config.get("master.auth_secret"), + { audience: this.sessionAud } + ); + + connector = this.activeConnectors.get(payload.sid); + if (!connector) { + throw new Error(); + } + + } catch (err) { + socket.send(JSON.stringify({ seq: null, type: "invalidate", data: {}})); + this.attachHandler(socket, req); + return; + } + + connector.continue(socket, data.last_seq); + return; + } + + if (this.stopAcceptingNewSessions) { + logger.verbose(`WsServer | closing ${req.socket.remoteAddress}, server is shutting down`); + wsRejectedConnectionsCounter.inc(); + socket.close(1001, "Shutting down"); + return; + } + + // Validate token + let user; + try { + if (type === "register_slave") { + let tokenPayload = jwt.verify( + data.token, + this.master.config.get("master.auth_secret"), + { audience: "slave" } + ); + + if (tokenPayload.slave !== data.id) { + throw new Error("missmatched slave id"); + } + + } else if (type === "register_control") { + let tokenPayload = jwt.verify( + data.token, + this.master.config.get("master.auth_secret"), + { audience: "user" } + ); + + user = this.master.userManager.users.get(tokenPayload.user); + if (!user) { + throw new Error("invalid user"); + } + if (tokenPayload.iat < user.tokenValidAfter) { + throw new Error("invalid token"); + } + user.checkPermission("core.control.connect"); + } + + } catch (err) { + logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`); + wsRejectedConnectionsCounter.inc(); + socket.close(4003, `Authentication failed: ${err.message}`); + return; + } + + let sessionId = this.nextSessionId; + this.nextSessionId += 1; + let sessionToken = jwt.sign( + { aud: this.sessionAud, sid: sessionId }, this.master.config.get("master.auth_secret") + ); + let heartbeatInterval = this.master.config.get("master.heartbeat_interval"); + let connector = new WsServerConnector(socket, sessionId, heartbeatInterval); + this.activeConnectors.set(sessionId, connector); + connector.on("close", () => { + this.activeConnectors.delete(connector); + }); + + let additionalReadyData = {}; + if (type === "register_slave") { + let connection = this.slaveConnections.get(data.id); + if (connection) { + logger.verbose(`WsServer | disconnecting existing connection for slave ${data.id}`); + connection.connector.setTimeout(15); // Slave connection is likely stalled + await connection.disconnect(1008, "Registered from another connection"); + } + + logger.verbose(`WsServer | registered slave ${data.id} version ${data.version}`); + connection = new SlaveConnection(data, connector, this.master); + connector.on("close", () => { + if (this.slaveConnections.get(data.id) === connection) { + this.slaveConnections.delete(data.id); + } + }); + this.slaveConnections.set(data.id, connection); + + + } else if (type === "register_control") { + logger.verbose(`WsServer | registered control from ${req.socket.remoteAddress}`); + let connection = new ControlConnection(data, connector, this.master, user); + connector.on("close", () => { + let index = this.controlConnections.indexOf(this); + if (index !== -1) { + this.controlConnections.splice(index, 1); + } + }); + this.controlConnections.push(connection); + additionalReadyData["account"] = { + name: user.name, + }; + } + + connector.ready(sessionToken, additionalReadyData); + } + +} + +module.exports = WsServer; diff --git a/packages/master/src/WsServerConnector.js b/packages/master/src/WsServerConnector.js new file mode 100644 index 00000000..721e2c89 --- /dev/null +++ b/packages/master/src/WsServerConnector.js @@ -0,0 +1,218 @@ +"use strict"; +const events = require("events"); + +const libLink = require("@clusterio/lib/link"); +const { logger } = require("@clusterio/lib/logging"); + + +/** + * Connector for master server connections + * + * @extends module:lib/link.WebSocketBaseConnector + * @alias module:master/src/WsServerConnector + */ +class WsServerConnector extends libLink.WebSocketBaseConnector { + constructor(socket, sessionId, heartbeatInterval) { + super(); + + this._socket = socket; + this._sessionId = sessionId; + this._heartbeatInterval = heartbeatInterval; + + // The following states are used in the server connector + // handshake: Waiting for client to (re)connect. + // connected: Connection is online + // closing: Connection is in the process of being closed. + // closed: Connection has been closed + this._state = "handshake"; + this._timeout = 15 * 60; + } + + _check(expectedStates) { + if (!expectedStates.includes(this._state)) { + throw new Error( + `Expected state ${expectedStates} but state is ${this._state}` + ); + } + } + + /** + * Send ready over the socket + * + * Sends the ready message over the socket to initiate the session. + * + * @param {string} sessionToken - + * the session token to send to the client. + * @param {Object=} additionalData - + * extra properties to send along the ready message. + */ + ready(sessionToken, additionalData) { + this._socket.send(JSON.stringify({ + seq: null, + type: "ready", + data: { + session_token: sessionToken, + heartbeat_interval: this._heartbeatInterval, + ...additionalData, + }, + })); + + this._state = "connected"; + this._attachSocketHandlers(); + this.emit("connect"); + } + + /** + * Continue connection with the given socket + * + * Terminates the current socket and contiunes the session over the + * socket given from the message sequence given. + * + * @param {module:net.Socket} socket - New socket to continue on. + * @param {number} lastSeq - The last message the client received. + */ + continue(socket, lastSeq) { + + // It's possible the previous connection hasn't closed yet due to a + // stale connection. If this is the case the connector will have to + // wait for the previous socket to close down before it can continue + // from a new socket, or there will be overlapping events from both. + if (["connected", "closing"].includes(this._state)) { + this._socket.close(4003, "Session Hijacked"); + + // Correctly waiting for the connector to close and then + // handling all the edge cases like multiple continues happening + // in parallel or connections going stale is far too difficult. + // Kill the connection here and let the client reconnect later + // when the connector is not held up by an active connection. + socket.close(1013, "Session Busy"); + socket.terminate(); + return; + } + + this._socket = socket; + + if (this._timeoutId) { + clearTimeout(this._timeoutId); + this._timeoutId = null; + } + + this._socket.send(JSON.stringify({ + seq: null, + type: "continue", + data: { + last_seq: this._lastReceivedSeq, + heartbeat_interval: this._heartbeatInterval, + }, + })); + + this._state = "connected"; + this._attachSocketHandlers(); + this._dropSendBufferSeq(lastSeq); + for (let message of this._sendBuffer) { + this._socket.send(JSON.stringify(message)); + } + this.emit("resume"); + } + + setTimeout(timeout) { + if (this._timeoutId) { + clearTimeout(this._timeoutId); + } + + this._timeoutId = setTimeout(() => { this._timedOut(); }, timeout * 1000); + this._timeout = timeout; + } + + _timedOut() { + logger.verbose("SOCKET | Connection timed out"); + this._timeoutId = null; + this._lastReceivedSeq = null; + this._sendBuffer.length = 0; + this._state = "closed"; + this.emit("close"); + this.emit("invalidate"); + } + + _attachSocketHandlers() { + this.startHeartbeat(); + + this._socket.on("close", (code, reason) => { + logger.verbose(`SOCKET | Close (code: ${code}, reason: ${reason})`); + if (this._state === "closing") { + this._lastReceivedSeq = null; + this._sendBuffer.length = 0; + this._state = "closed"; + this.emit("close"); + + if (this._timeoutId) { + clearTimeout(this._timeoutId); + this._timeoutId = null; + } + + } else { + this._state = "handshake"; + this.emit("drop"); + this._timeoutId = setTimeout(() => { this._timedOut(); }, this._timeout * 1000); + } + + this.stopHeartbeat(); + }); + + this._socket.on("error", err => { + // It's assumed that close is always called by ws + logger.verbose("SOCKET | Error:", err); + }); + + this._socket.on("open", () => { + logger.verbose("SOCKET | Open"); + }); + this._socket.on("ping", data => { + logger.verbose(`SOCKET | Ping (data: ${data}`); + }); + this._socket.on("pong", data => { + logger.verbose(`SOCKET | Pong (data: ${data}`); + }); + + // Handle messages + this._socket.on("message", data => { + let message = JSON.parse(data); + if (["connected", "closing"].includes(this._state)) { + if (message.seq !== null) { + this._lastReceivedSeq = message.seq; + } + + if (message.type === "heartbeat") { + this._processHeartbeat(message); + + } else { + this.emit("message", message); + } + + } else { + throw new Error(`Received message in unexpected state ${this._state}`); + } + }); + } + + /** + * Close the connection with the given reason. + * + * Sends a close frame and disconnects the connector. + * + * @param {number} code - WebSocket close code. + * @param {string} reason - WebSocket close reason. + */ + async close(code, reason) { + if (this._state === "closed") { + return; + } + + this.stopHeartbeat(); + this._state = "closing"; + this._socket.close(code, reason); + await events.once(this, "close"); + } +} + +module.exports = WsServerConnector; diff --git a/packages/master/src/metrics.js b/packages/master/src/metrics.js new file mode 100644 index 00000000..eb753828 --- /dev/null +++ b/packages/master/src/metrics.js @@ -0,0 +1,9 @@ +// Shared metrics +"use strict"; +const libPrometheus = require("@clusterio/lib/prometheus"); + +exports.endpointHitCounter = new libPrometheus.Counter( + "clusterio_master_http_endpoint_hits_total", + "How many requests a particular HTTP endpoint has gotten", + { labels: ["route"] } +); diff --git a/packages/master/src/routes.js b/packages/master/src/routes.js index c72ab970..8ab2cba0 100644 --- a/packages/master/src/routes.js +++ b/packages/master/src/routes.js @@ -1,6 +1,158 @@ "use strict"; +const fs = require("fs-extra"); +const JSZip = require("jszip"); +const jwt = require("jsonwebtoken"); +const path = require("path"); + +const libHelpers = require("@clusterio/lib/helpers"); +const libLink = require("@clusterio/lib/link"); +const libPlugin = require("@clusterio/lib/plugin"); +const libPrometheus = require("@clusterio/lib/prometheus"); + +const { endpointHitCounter } = require("./metrics"); + + +// Merges samples from sourceResult to destinationResult +function mergeSamples(destinationResult, sourceResult) { + let receivedSamples = new Map(sourceResult.samples); + for (let [suffix, suffixSamples] of destinationResult.samples) { + if (receivedSamples.has(suffix)) { + suffixSamples.push(...receivedSamples.get(suffix)); + receivedSamples.delete(suffix); + } + } + + for (let entry of receivedSamples) { + sourceResult.samples.push(entry); + } +} + +// Prometheus polling endpoint +async function getMetrics(req, res, next) { + endpointHitCounter.labels(req.route.path).inc(); + + let results = []; + let pluginResults = await libPlugin.invokeHook(req.app.locals.master.plugins, "onMetrics"); + for (let metricIterator of pluginResults) { + for await (let metric of metricIterator) { + results.push(metric); + } + } + + let requests = []; + let timeout = req.app.locals.master.config.get("master.metrics_timeout") * 1000; + for (let slaveConnection of req.app.locals.master.wsServer.slaveConnections.values()) { + requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), timeout, null)); + } + + for await (let result of await libPrometheus.defaultRegistry.collect()) { + results.push(result); + } + + let resultMap = new Map(); + for (let response of await Promise.all(requests)) { + if (!response) { + // TODO: Log timeout occured? + continue; + } + + for (let result of response.results) { + if (!resultMap.has(result.metric.name)) { + resultMap.set(result.metric.name, result); + + } else { + // Merge metrics received by multiple slaves + mergeSamples(resultMap.get(result.metric.name), result); + } + } + } + + for (let result of resultMap.values()) { + results.push(libPrometheus.deserializeResult(result)); + } + + + let text = await libPrometheus.exposition(results); + res.set("Content-Type", libPrometheus.exposition.contentType); + res.send(text); +} + +function getPlugins(req, res) { + let plugins = []; + for (let pluginInfo of req.app.locals.master.pluginInfos) { + let name = pluginInfo.name; + let loaded = req.app.locals.master.plugins.has(name); + let enabled = loaded && req.app.locals.master.config.group(pluginInfo.name).get("load_plugin"); + plugins.push({ name, version: pluginInfo.version, enabled, loaded }); + } + res.send(plugins); +} + +function validateSlaveToken(req, res, next) { + let token = req.header("x-access-token"); + if (!token) { + res.sendStatus(401); + return; + } + + try { + jwt.verify(token, req.app.locals.master.config.get("master.auth_secret"), { audience: "slave" }); + + } catch (err) { + res.sendStatus(401); + return; + } + + next(); +} + +// Handle an uploaded export package. +async function uploadExport(req, res) { + endpointHitCounter.labels(req.route.path).inc(); + if (req.get("Content-Type") !== "application/zip") { + res.sendStatus(415); + return; + } + + let data = []; + for await (let chunk of req) { + data.push(chunk); + } + data = Buffer.concat(data); + let zip = await JSZip.loadAsync(data); + data = null; + + // This is hardcoded to prevent path expansion attacks + let exportFiles = [ + "export/item-spritesheet.png", + "export/item-metadata.json", + "export/locale.json", + ]; + + for (let filePath of exportFiles) { + let file = zip.file(filePath); + if (!file) { + continue; + } + + let name = path.posix.basename(filePath); + await fs.outputFile(path.join("static", "export", name), await file.async("nodebuffer")); + } + + res.sendStatus(200); +} + +function addRouteHandlers(app) { + app.get("/metrics", (req, res, next) => getMetrics(req, res, next).catch(next)); + app.get("/api/plugins", getPlugins); + app.put("/api/upload-export", + validateSlaveToken, + (req, res, next) => uploadExport(req, res).catch(next) + ); +} + // Routes used in the web interface and served by the master server -module.exports = [ +const webRoutes = [ "/", "/master", "/slaves", @@ -14,3 +166,8 @@ module.exports = [ "/plugins", "/plugins/:name/view", ]; + +module.exports = { + addRouteHandlers, + webRoutes, +}; diff --git a/packages/slave/slave.js b/packages/slave/slave.js index a1217226..2ee280b2 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -224,7 +224,7 @@ class Instance extends libLink.Link { /** * Current state of the instance * - * One of stopped, starting, running, creating_save and exporting_data + * One of stopped, starting, running, stopping, creating_save and exporting_data * * @returns {string} instance status. */ @@ -623,6 +623,7 @@ class Instance extends libLink.Link { */ async stop() { this._running = false; + this.notifyStatus("stopping"); // XXX this needs more thought to it if (this.server._state === "running") { diff --git a/plugins/player_auth/master.js b/plugins/player_auth/master.js index 5c491014..ba94fc16 100644 --- a/plugins/player_auth/master.js +++ b/plugins/player_auth/master.js @@ -3,6 +3,7 @@ */ "use strict"; const crypto = require("crypto"); +const express = require("express"); const util = require("util"); const jwt = require("jsonwebtoken"); @@ -46,7 +47,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { this.metrics.endpointHitCounter.labels(req.route.path).inc(); let servers = []; - for (let instance of this.master.db.instances.values()) { + for (let instance of this.master.instances.values()) { if (instance.status === "running" && instance.config.get("player_auth.load_plugin")) { servers.push(instance.config.get("factorio.settings")["name"] || "unnamed server"); } @@ -54,12 +55,12 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { res.send(servers); }); - this.master.app.post("/api/player_auth/player_code", (req, res, next) => { + this.master.app.post("/api/player_auth/player_code", express.json(), (req, res, next) => { this.metrics.endpointHitCounter.labels(req.route.path).inc(); this.handlePlayerCode(req, res).catch(next); }); - this.master.app.post("/api/player_auth/verify", (req, res, next) => { + this.master.app.post("/api/player_auth/verify", express.json(), (req, res, next) => { this.metrics.endpointHitCounter.labels(req.route.path).inc(); this.handleVerify(req, res).catch(next); }); @@ -142,7 +143,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { for (let [player, entry] of this.players) { if (entry.playerCode === playerCode && entry.expires > Date.now()) { if (entry.verifyCode === verifyCode) { - let user = this.master.db.users.get(player); + let user = this.master.userManager.users.get(player); if (!user) { res.send({ error: true, message: "invalid user" }); return; diff --git a/plugins/player_auth/package.json b/plugins/player_auth/package.json index 8d7a141e..864a6242 100644 --- a/plugins/player_auth/package.json +++ b/plugins/player_auth/package.json @@ -20,6 +20,7 @@ "@clusterio/lib": "^2.0.0-alpha.0" }, "dependencies": { + "express": "^4.16.2", "jsonwebtoken": "^8.2.1" }, "devDependencies": { diff --git a/plugins/player_auth/test/plugin.js b/plugins/player_auth/test/plugin.js index 9fd42a73..0f7f6983 100644 --- a/plugins/player_auth/test/plugin.js +++ b/plugins/player_auth/test/plugin.js @@ -53,7 +53,7 @@ describe("player_auth", function() { describe("/api/player_auth/servers", function() { it("should return a list of running servers with player_auth loaded", async function() { function addInstance(id, status, load, name) { - masterPlugin.master.db.instances.set(id, { + masterPlugin.master.instances.set(id, { config: { get(field) { if (field === "player_auth.load_plugin") { @@ -78,7 +78,7 @@ describe("player_auth", function() { }); assert.deepEqual(result.body, ["running loaded", "unnamed server"]); for (let id of [1, 2, 3, 4, 5]) { - masterPlugin.master.db.instances.delete(id); + masterPlugin.master.instances.delete(id); } }); }); diff --git a/plugins/subspace_storage/master.js b/plugins/subspace_storage/master.js index 0472a379..18ffc6dd 100644 --- a/plugins/subspace_storage/master.js +++ b/plugins/subspace_storage/master.js @@ -137,8 +137,8 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { } } else { - let instanceConfig = this.master.db.instances.get(instanceId); - let instanceName = instanceConfig ? instanceConfig.get("instance.name") : "unkonwn"; + let instance = this.master.instances.get(instanceId); + let instanceName = instance ? instance.config.get("instance.name") : "unkonwn"; // use fancy neural net to calculate a "fair" dole division rate. if (method === "neural_dole") { diff --git a/test/integration/index.js b/test/integration/index.js index 0d9b8bf1..d15bb15e 100644 --- a/test/integration/index.js +++ b/test/integration/index.js @@ -156,7 +156,7 @@ before(async function() { await exec("node ../../packages/master bootstrap create-ctl-config test"); await exec("node ../../packages/ctl control-config set control.tls_ca ../../test/file/tls/cert.pem"); - masterProcess = await spawn("master:", "node ../../packages/master run", /All plugins loaded/); + masterProcess = await spawn("master:", "node ../../packages/master run", /Started master/); await execCtl("slave create-config --id 4 --name slave --generate-token"); await exec(`node ../../packages/slave config set slave.factorio_directory ${path.join("..", "..", "factorio")}`); diff --git a/test/mock.js b/test/mock.js index 7536771c..5d11a8e0 100644 --- a/test/mock.js +++ b/test/mock.js @@ -2,13 +2,14 @@ const events = require("events"); const http = require("http"); const express = require("express"); -const bodyParser = require("body-parser"); const libLink = require("@clusterio/lib/link"); const libPlugin = require("@clusterio/lib/plugin"); const libUsers = require("@clusterio/lib/users"); const libPrometheus = require("@clusterio/lib/prometheus"); +const UserManager = require("@clusterio/master/src/UserManager"); + class MockLogger { child() { return this; } @@ -149,7 +150,6 @@ class MockControl extends libLink.Link { class MockMaster { constructor() { this.app = express(); - this.app.use(bodyParser.json({ limit: "10mb" })); this.mockConfigEntries = new Map([ ["master.external_address", "test"], ["master.auth_secret", "TestSecretDoNotUse"], @@ -163,22 +163,19 @@ class MockMaster { }, }; - let roles = new Map([ + this.userManager = new UserManager(); + this.userManager.roles = new Map([ [0, new libUsers.Role({ id: 0, name: "Admin", description: "admin", permissions: ["core.admin"] })], [1, new libUsers.Role({ id: 1, name: "Player", description: "player", permissions: [] })], ]); - roles.get(1).grantDefaultPermissions(); + this.userManager.roles.get(1).grantDefaultPermissions(); - let users = new Map([ - ["test", new libUsers.User({ name: "test", roles: [0, 1] }, roles)], - ["player", new libUsers.User({ name: "player", roles: [1] }, roles)], + this.userManager.users = new Map([ + ["test", new libUsers.User({ name: "test", roles: [0, 1] }, this.userManager.roles)], + ["player", new libUsers.User({ name: "player", roles: [1] }, this.userManager.roles)], ]); - this.db = { - instances: new Map(), - slaves: new Map(), - roles, - users, - }; + this.instances = new Map(); + this.slaves = new Map(); } getMasterUrl() { diff --git a/test/package.json b/test/package.json index 74e3207b..0c9d52dd 100644 --- a/test/package.json +++ b/test/package.json @@ -9,7 +9,6 @@ "@clusterio/lib": "2.0.0-alpha.5", "@clusterio/master": "2.0.0-alpha.5", "@clusterio/slave": "2.0.0-alpha.5", - "body-parser": "^1.18.2", "express": "^4.16.2", "fs-extra": "^8.1.0", "jsonwebtoken": "^8.2.1",