aperx/factorio-clusterio
1
0 Fork 0
mirror of https://github.com/clusterio/clusterio.git synced 2026-11-12 15:04:52 +09:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #495 from Luciole0x/master

Browse Source 
Transition controller package to TypeScript.
This commit is contained in:
Hornwitser authored and GitHub committed 2023-08-15 17:11:26 +02:00
2 parents b72fa25d3f e59027f557
commit e51a644788
28 files changed
+1164 -884
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/controller/controller.js → packages/controller/controller.ts
Executable → Regular
+113 -66
View File
@@ -14,27 +14,28 @@
* npx clusteriocontroller run
*/
"use strict";
const path = require("path");
const fs = require("fs-extra");
const crypto = require("crypto");
const setBlocking = require("set-blocking");
const yargs = require("yargs");
const util = require("util");
const winston = require("winston");
require("winston-daily-rotate-file");
const jwt = require("jsonwebtoken");
import path from "path";
import fs from "fs-extra";
import crypto from "crypto";
import setBlocking from "set-blocking";
import yargs from "yargs";
import util from "util";
import winston from "winston";
import DailyRotateFile from "winston-daily-rotate-file";
import jwt from "jsonwebtoken";
// homebrew modules
const lib = require("@clusterio/lib");
import * as lib from "@clusterio/lib";
const { ConsoleTransport, levels, logger } = lib;
const Controller = require("./src/Controller");
const UserManager = require("./src/UserManager");
const version = require("./package").version;
import Controller from "./src/Controller";
import UserManager from "./src/UserManager";
const version = require("./package.json").version;
// globals
let controller;
let controller: Controller;
void new lib.Gauge(
@@ -42,7 +43,7 @@ void new lib.Gauge(
"Mapping of Host ID to name",
{
labels: ["host_id", "host_name"],
callback: function(gauge) {
callback: function(gauge:lib.Gauge) {
gauge.clear();
if (!controller || !controller.hosts) {
return;
@@ -62,7 +63,7 @@ void new lib.Gauge(
"Mapping of Instance ID to name and host",
{
labels: ["instance_id", "instance_name", "host_id"],
callback: function(gauge) {
callback: function(gauge:lib.Gauge) {
gauge.clear();
if (!controller || !controller.instances) {
return;
@@ -81,19 +82,26 @@ void new lib.Gauge(
void new lib.Gauge(
"clusterio_controller_websocket_active_connections",
"How many WebSocket connections are currently open to the controller",
{ callback: function(gauge) { gauge.set(controller.wsServer.activeConnectors.size); }}
{ callback: function(gauge:lib.Gauge) {
gauge.set(controller.wsServer.activeConnectors.size);
}
}
);
void new lib.Gauge(
"clusterio_controller_active_hosts",
"How many hosts are currently connected to the controller",
{ callback: function(gauge) { gauge.set(controller.wsServer.hostConnections.size); }}
{ callback: function(gauge:lib.Gauge) {
gauge.set(controller.wsServer.hostConnections.size);
}
}
);
void new lib.Gauge(
"clusterio_controller_connected_clients_count", "How many clients are currently connected to this controller",
"clusterio_controller_connected_clients_count",
"How many clients are currently connected to this controller",
{
labels: ["type"], callback: async function(gauge) {
labels: ["type"], callback: async function(gauge:lib.Gauge) {
gauge.labels("host").set(controller.wsServer.hostConnections.size);
gauge.labels("control").set(controller.wsServer.controlConnections.size);
},
@@ -101,7 +109,10 @@ void new lib.Gauge(
);
async function handleBootstrapCommand(args, controllerConfig) {
async function handleBootstrapCommand(
args: any,
controllerConfig: lib.ControllerConfig
): Promise<void> {
let subCommand = args._[1];
let userManager = new UserManager(controllerConfig);
await userManager.load(path.join(controllerConfig.get("controller.database_directory"), "users.json"));
@@ -166,18 +177,31 @@ async function handleBootstrapCommand(args, controllerConfig) {
}
}
async function initialize() {
let parameters = {
args: null,
shouldRun: false,
clusterLogger: null,
pluginInfos: null,
controllerConfigPath: null,
controllerConfig: null,
};
export interface ControllerArgs {
[x: string]: unknown;
logLevel: string;
logDirectory: string;
pluginList: string;
devPlugin?: (string | number)[] | undefined;
dev?: boolean;
config: string;
_: (string | number)[];
$0: string;
}
interface InitializeParameters {
args: ControllerArgs;
shouldRun: boolean;
clusterLogger: winston.Logger;
pluginInfos: any[] | null;
controllerConfigPath: string | null;
controllerConfig: lib.ControllerConfig | null;
}
async function initialize(): Promise<InitializeParameters> {
// argument parsing
parameters.args = yargs
const args = await yargs
.scriptName("controller")
.usage("$0 <command> [options]")
.option("log-level", {
@@ -232,45 +256,46 @@ async function initialize() {
;
// Combined log stream of the whole cluster.
parameters.clusterLogger = winston.createLogger({
const clusterLogger = winston.createLogger({
format: winston.format.json(),
level: "verbose",
levels,
});
parameters.clusterLogger.add(new winston.transports.DailyRotateFile({
clusterLogger.add(new DailyRotateFile({
filename: "cluster-%DATE%.log",
utc: true,
dirname: path.join(parameters.args.logDirectory, "cluster"),
dirname: path.join(args.logDirectory, "cluster"),
}));
// Log stream for the controller.
logger.add(new winston.transports.DailyRotateFile({
logger.add(new DailyRotateFile({
format: winston.format.json(),
filename: "controller-%DATE%.log",
dirname: path.join(parameters.args.logDirectory, "controller"),
dirname: path.join(args.logDirectory, "controller"),
}));
logger.add(new winston.transports.Stream({
stream: parameters.clusterLogger,
stream: clusterLogger,
}));
if (parameters.args.logLevel !== "none") {
if (args.logLevel !== "none") {
logger.add(new ConsoleTransport({
level: parameters.args.logLevel,
level: args.logLevel,
format: new lib.TerminalFormat(),
}));
}
lib.handleUnhandledErrors(logger);
lib.handleUnhandledErrors();
let command = parameters.args._[0];
let command = args._[0];
let shouldRun = false;
if (command === "run") {
logger.info(`Starting Clusterio controller ${version}`);
parameters.shouldRun = true;
shouldRun = true;
}
logger.info(`Loading available plugins from ${parameters.args.pluginList}`);
logger.info(`Loading available plugins from ${args.pluginList}`);
let pluginList = new Map();
try {
pluginList = new Map(JSON.parse(await fs.readFile(parameters.args.pluginList)));
} catch (err) {
pluginList = new Map(JSON.parse(await fs.readFile(args.pluginList, { encoding: 'utf8' })));
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
@@ -278,52 +303,67 @@ async function initialize() {
// If the command is plugin management we don't try to load plugins
if (command === "plugin") {
await lib.handlePluginCommand(parameters.args, pluginList, parameters.args.pluginList);
return parameters;
await lib.handlePluginCommand(args, pluginList, args.pluginList);
return {
args,
shouldRun,
clusterLogger,
pluginInfos: null,
controllerConfigPath: null,
controllerConfig: null,
};
}
logger.info("Loading Plugin info");
parameters.pluginInfos = await lib.loadPluginInfos(pluginList);
lib.registerPluginMessages(parameters.pluginInfos);
lib.registerPluginConfigGroups(parameters.pluginInfos);
const pluginInfos = await lib.loadPluginInfos(pluginList);
lib.registerPluginMessages(pluginInfos);
lib.registerPluginConfigGroups(pluginInfos);
lib.finalizeConfigs();
parameters.controllerConfigPath = parameters.args.config;
logger.info(`Loading config from ${parameters.controllerConfigPath}`);
parameters.controllerConfig = new lib.ControllerConfig("controller");
const controllerConfigPath = args.config;
logger.info(`Loading config from ${controllerConfigPath}`);
const controllerConfig = new lib.ControllerConfig("controller");
try {
await parameters.controllerConfig.load(JSON.parse(await fs.readFile(parameters.controllerConfigPath)));
let fileData = await fs.readFile(controllerConfigPath, { encoding: 'utf8' })
await controllerConfig.load(JSON.parse(fileData));
} catch (err) {
} catch (err: any) {
if (err.code === "ENOENT") {
logger.info("Config not found, initializing new config");
await parameters.controllerConfig.init();
await controllerConfig.init();
} else {
throw new lib.StartupError(`Failed to load ${parameters.controllerConfigPath}: ${err.message}`);
throw new lib.StartupError(`Failed to load ${controllerConfigPath}: ${err.message}`);
}
}
if (!parameters.controllerConfig.get("controller.auth_secret")) {
if (!controllerConfig.get("controller.auth_secret")) {
logger.info("Generating new controller authentication secret");
let asyncRandomBytes = util.promisify(crypto.randomBytes);
let bytes = await asyncRandomBytes(256);
parameters.controllerConfig.set("controller.auth_secret", bytes.toString("base64"));
controllerConfig.set("controller.auth_secret", bytes.toString("base64"));
await lib.safeOutputFile(
parameters.controllerConfigPath, JSON.stringify(parameters.controllerConfig.serialize(), null, 4)
controllerConfigPath, JSON.stringify(controllerConfig.serialize(), null, 4)
);
}
if (command === "config") {
await lib.handleConfigCommand(
parameters.args, parameters.controllerConfig, parameters.controllerConfigPath
args, controllerConfig, controllerConfigPath
);
} else if (command === "bootstrap") {
await handleBootstrapCommand(parameters.args, parameters.controllerConfig);
await handleBootstrapCommand(args, controllerConfig);
}
return parameters;
return {
args,
clusterLogger,
pluginInfos,
controllerConfigPath,
controllerConfig,
shouldRun,
}
}
async function startup() {
@@ -331,8 +371,15 @@ async function startup() {
// and as the process name in ps/top on linux.
process.title = "clusteriocontroller";
let { args, shouldRun, clusterLogger, pluginInfos, controllerConfigPath, controllerConfig } = await initialize();
if (!shouldRun) {
let {
args,
shouldRun,
clusterLogger,
pluginInfos,
controllerConfigPath,
controllerConfig
} = await initialize();
if (!shouldRun || !pluginInfos || !controllerConfigPath || !controllerConfig) {
return;
}
packages/controller/package.json
+14 -3
View File
@@ -4,13 +4,14 @@
"author": "Hornwitser",
"license": "MIT",
"version": "2.0.0-alpha.13",
"main": "controller.js",
"main": "dist/controller/controller.js",
"scripts": {
"prepare": "webpack-cli --env production",
"prepare": "webpack-cli --env production && pnpm build",
"build": "tsc --build",
"test": "echo \"Error: no test specified\" && exit 1"
},
"bin": {
"clusteriocontroller": "./controller.js"
"clusteriocontroller": "./dist/controller/controller.js"
},
"keywords": [
"factorio"
@@ -35,9 +36,19 @@
},
"devDependencies": {
"@clusterio/web_ui": "^2.0.0-alpha.13",
"@types/busboy": "^1.5.0",
"@types/compression": "^1.7.2",
"@types/express": "^4.17.17",
"@types/fs-extra": "^11.0.1",
"@types/jsonwebtoken": "^9.0.2",
"@types/node": "^20.4.5",
"@types/set-blocking": "^2.0.0",
"@types/ws": "^8.5.5",
"@types/yargs": "^17.0.24",
"react": "^18.2.0",
"react-dom": "^18.2.0",
"react-router-dom": "^6.4.14",
"typescript": "^5.1.6",
"webpack": "^5.88.2",
"webpack-dev-middleware": "^6.1.1",
"webpack-merge": "^5.9.0"
packages/controller/src/BaseConnection.js
-69
View File
@@ -1,69 +0,0 @@
"use strict";
const lib = require("@clusterio/lib");
const { logger } = lib;
const ControllerRouter = require("./ControllerRouter");
/**
* Base class for controller connections
*
* @extends module:lib.Link
* @alias module:controller/src/BaseConnection
*/
class BaseConnection extends lib.Link {
constructor(connector, controller) {
/** @member {module:controller/src/WsServerConnector} module:controller/src/BaseConnection#connector */
super(connector);
this.router = new ControllerRouter(controller);
this._controller = controller;
for (let [Request, handler] of controller._registeredRequests) { this.handle(Request, handler); }
for (let [Request, handler] of controller._fallbackedRequests) { this.fallbackRequest(Request, handler); }
for (let [Event, handler] of controller._registeredEvents) { this.handle(Event, handler); }
for (let [Event, handler] of controller._snoopedEvents) { this.snoopEvent(Event, handler); }
this.handle(lib.ModPackGetRequest, this.handleModPackGetRequest.bind(this));
this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this));
}
async disconnect(code, reason) {
try {
await this.connector.disconnect();
} catch (err) {
if (!(err instanceof lib.SessionLost)) {
logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`);
}
}
}
/**
* True if the link is connected, not in the dropped state and not in
* the process of disconnecting.
* @type {boolean}
*/
get connected() {
return this.connector.connected;
}
async handleModPackGetRequest(request) {
let { id } = request;
let modPack = this._controller.modPacks.get(id);
if (!modPack) {
throw new lib.RequestError(`Mod pack with ID ${id} does not exist`);
}
return modPack;
}
async handleModPackGetDefaultRequest() {
let id = this._controller.config.get("controller.default_mod_pack_id");
if (id === null) {
throw new lib.RequestError("Default mod pack not set on controller");
}
let modPack = this._controller.modPacks.get(id);
if (!modPack) {
throw new lib.RequestError(`Default mod pack configured (${id}) does not exist`);
}
return modPack;
}
}
module.exports = BaseConnection;
packages/controller/src/BaseConnection.ts
+71
View File
@@ -0,0 +1,71 @@
import type Controller from "./Controller";
import type WsServerConnector from "./WsServerConnector";
import * as lib from "@clusterio/lib";
const { logger } = lib;
import ControllerRouter from "./ControllerRouter";
/**
* Base class for controller connections
*
* @extends module:lib.Link
* @alias module:controller/src/BaseConnection
*/
export default class BaseConnection extends lib.Link {
declare connector: lib.WebSocketBaseConnector;
constructor(
connector: lib.WebSocketBaseConnector,
public _controller: Controller
) {
super(connector);
this.router = new ControllerRouter(this._controller);
for (let [Request, handler] of this._controller._registeredRequests) { this.handle(Request, handler); }
for (let [Request, handler] of this._controller._fallbackedRequests) { this.fallbackRequest(Request, handler); }
for (let [Event, handler] of this._controller._registeredEvents) { this.handle(Event, handler); }
for (let [Event, handler] of this._controller._snoopedEvents) { this.snoopEvent(Event, handler); }
this.handle(lib.ModPackGetRequest, this.handleModPackGetRequest.bind(this));
this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this));
}
async disconnect(code: number, reason: string) {
try {
await this.connector.disconnect();
} catch (err: any) {
if (!(err instanceof lib.SessionLost)) {
logger.error(`"Unexpected error preparing disconnect:\n${err.stack}`);
}
}
}
/**
* True if the link is connected, not in the dropped state and not in
* the process of disconnecting.
*/
get connected(): boolean {
return this.connector.connected;
}
async handleModPackGetRequest(request: lib.ModPackGetRequest): Promise<lib.ModPack> {
let { id } = request;
let modPack = this._controller.modPacks!.get(id);
if (!modPack) {
throw new lib.RequestError(`Mod pack with ID ${id} does not exist`);
}
return modPack;
}
async handleModPackGetDefaultRequest(): Promise<lib.ModPack> {
let id = this._controller.config.get("controller.default_mod_pack_id");
if (id === null) {
throw new lib.RequestError("Default mod pack not set on controller");
}
let modPack = this._controller.modPacks!.get(id);
if (!modPack) {
throw new lib.RequestError(`Default mod pack configured (${id}) does not exist`);
}
return modPack;
}
}
packages/controller/src/ControlConnection.js → packages/controller/src/ControlConnection.ts
+210 -171
View File
@@ -1,15 +1,20 @@
"use strict";
const events = require("events");
const fs = require("fs-extra");
const path = require("path");
import type WsServerConnector from "./WsServerConnector";
import type { HostInfo } from "./HostConnection";
const lib = require("@clusterio/lib");
import events from "events";
import fs from "fs-extra";
import path from "path";
import * as lib from "@clusterio/lib";
const { logFilter, logger } = lib;
const BaseConnection = require("./BaseConnection");
const routes = require("./routes");
import BaseConnection from "./BaseConnection";
import * as routes from "./routes";
import Controller from "./Controller";
import { promises } from "dns";
import InstanceInfo from "./InstanceInfo";
const strcmp = new Intl.Collator(undefined, { numerice: "true", sensitivity: "base" }).compare;
const strcmp = new Intl.Collator(undefined, { numeric: true, sensitivity: "base" }).compare;
const queryLogTime = new lib.Summary(
"clusterio_controller_query_log_duration_seconds",
@@ -22,66 +27,49 @@ const queryLogTime = new lib.Summary(
* @extends module:controller/src/BaseConnection
* @alias module:controller/src/ControlConnection
*/
class ControlConnection extends BaseConnection {
constructor(registerData, connector, controller, user, id) {
export default class ControlConnection extends BaseConnection {
private _agent: string;
private _version: string;
hostSubscriptions: { all:boolean, hostIds:number[] } = { all:false, hostIds:[] };
instanceSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] };
saveListSubscriptions: { all:boolean, instanceIds:number[] } = { all:false, instanceIds:[] };
modPackSubscriptions: { all:boolean, modPackIds:number[] } = { all:false, modPackIds:[] };
modSubscriptions: { all:boolean, modNames:string[] } = { all:false, modNames:[] };
userSubscriptions: { all:boolean, names:string[]} = { all:false, names:[] };
logTransport: lib.LinkTransport | null = null;
logSubscriptions: {
all: boolean,
controller: boolean,
hostIds: number[],
instanceIds: number[],
} = {
all: false,
controller: false,
hostIds: [],
instanceIds: [],
};
ws_dumper: ((...args: any[]) => void) | null = null;
declare connector: WsServerConnector;
constructor(
registerData: { agent:string, version:string },
connector: WsServerConnector,
controller: Controller,
public user: lib.User, // The user making this connection.
public id: number
) {
super(connector, controller);
this._agent = registerData.agent;
this._version = registerData.version;
this.id = id;
/**
* The user making this connection.
* @type {module:lib.User}
*/
this.user = user;
this.hostSubscriptions = {
all: false,
hostIds: [],
};
this.instanceSubscriptions = {
all: false,
instanceIds: [],
};
this.saveListSubscriptions = {
all: false,
instanceIds: [],
};
this.modPackSubscriptions = {
all: false,
modPackIds: [],
};
this.modSubscriptions = {
all: false,
modNames: [],
};
this.userSubscriptions = {
all: false,
names: [],
};
this.logTransport = null;
this.logSubscriptions = {
all: false,
controller: false,
hostIds: [],
instanceIds: [],
};
this.ws_dumper = null;
this.connector.on("connect", () => {
this.connector._socket.clusterio_ignore_dump = Boolean(this.ws_dumper);
this.connector._socket!.clusterio_ignore_dump = Boolean(this.ws_dumper);
});
this.connector.on("close", () => {
if (this.logTransport) {
this.logTransport = null;
logger.remove(this.logTransport);
this.logTransport = null;
}
if (this.ws_dumper) {
this._controller.debugEvents.off("message", this.ws_dumper);
@@ -91,7 +79,10 @@ class ControlConnection extends BaseConnection {
for (let event of ["connect", "drop", "resume", "close"]) {
this.connector.on(event, () => {
for (let controllerPlugin of this._controller.plugins.values()) {
controllerPlugin.onControlConnectionEvent(this, event);
controllerPlugin.onControlConnectionEvent(
this,
event as "connect" | "drop" | "resume" | "close"
);
}
});
}
@@ -105,26 +96,17 @@ class ControlConnection extends BaseConnection {
this.handle(lib.HostConfigCreateRequest, this.handleHostConfigCreateRequest.bind(this));
this.handle(lib.InstanceDetailsGetRequest, this.handleInstanceDetailsGetRequest.bind(this));
this.handle(lib.InstanceDetailsListRequest, this.handleInstanceDetailsListRequest.bind(this));
this.handle(
lib.InstanceDetailsSetSubscriptionsRequest, this.handleInstanceDetailsSetSubscriptionsRequest.bind(this)
);
this.handle(lib.InstanceDetailsSetSubscriptionsRequest, this.handleInstanceDetailsSetSubscriptionsRequest.bind(this));
this.handle(lib.InstanceCreateRequest, this.handleInstanceCreateRequest.bind(this));
this.handle(lib.InstanceDeleteRequest, this.handleInstanceDeleteRequest.bind(this));
this.handle(lib.InstanceConfigGetRequest, this.handleInstanceConfigGetRequest.bind(this));
this.handle(lib.InstanceConfigSetFieldRequest, this.handleInstanceConfigSetFieldRequest.bind(this));
this.handle(lib.InstanceConfigSetPropRequest, this.handleInstanceConfigSetPropRequest.bind(this));
this.handle(lib.InstanceAssignRequest, this.handleInstanceAssignRequest.bind(this));
this.handle(
lib.InstanceSetSaveListSubscriptionsRequest,
this.handleInstanceSetSaveListSubscriptionsRequest.bind(this)
);
this.handle(
lib.InstanceRenameSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)
);
this.handle(lib.InstanceSetSaveListSubscriptionsRequest, this.handleInstanceSetSaveListSubscriptionsRequest.bind(this));
this.handle(lib.InstanceRenameSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
this.handle(lib.InstanceCopySaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
this.handle(
lib.InstanceDeleteSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller)
);
this.handle(lib.InstanceDeleteSaveRequest, this._controller.sendToHostByInstanceId.bind(this._controller));
this.handle(lib.InstanceDownloadSaveRequest, this.handleInstanceDownloadSaveRequest.bind(this));
this.handle(lib.InstanceTransferSaveRequest, this.handleInstanceTransferSaveRequest.bind(this));
this.handle(lib.ModPackListRequest, this.handleModPackListRequest.bind(this));
@@ -144,9 +126,7 @@ class ControlConnection extends BaseConnection {
this.handle(lib.RoleListRequest, this.handleRoleListRequest.bind(this));
this.handle(lib.RoleCreateRequest, this.handleRoleCreateRequest.bind(this));
this.handle(lib.RoleUpdateRequest, this.handleRoleUpdateRequest.bind(this));
this.handle(
lib.RoleGrantDefaultPermissionsRequest, this.handleRoleGrantDefaultPermissionsRequest.bind(this)
);
this.handle(lib.RoleGrantDefaultPermissionsRequest, this.handleRoleGrantDefaultPermissionsRequest.bind(this));
this.handle(lib.RoleDeleteRequest, this.handleRoleDeleteRequest.bind(this));
this.handle(lib.UserGetRequest, this.handleUserGetRequest.bind(this));
this.handle(lib.UserListRequest, this.handleUserListRequest.bind(this));
@@ -161,29 +141,35 @@ class ControlConnection extends BaseConnection {
this.handle(lib.DebugDumpWsRequest, this.handleDebugDumpWsRequest.bind(this));
}
validateIngress(message) {
validateIngress(message: lib.MessageRequest | lib.MessageEvent) {
let origin = this.connector.dst;
if (origin.type !== message.src.type || origin.id !== message.src.id) {
throw new lib.InvalidMessage(`Received message with invalid src ${message.src} from ${origin}`);
}
}
validatePermission(message, entry) {
validatePermission(
message: lib.MessageRequest | lib.MessageEvent,
entry: lib.RequestEntry | lib.EventEntry
) {
try {
this.checkPermission(message, entry);
} catch (err) {
} catch (err: any) {
this.connector.sendResponseError(new lib.ResponseError(err.message, err.code), message.src);
logger.audit(`Permission denied for ${message.name} by ${this.user.name} from ${this.connector.dst}`);
throw err;
}
}
checkPermission(message, entry) {
checkPermission(
message: lib.MessageRequest | lib.MessageEvent,
entry: lib.RequestEntry | lib.EventEntry
) {
let permission;
if (message.type === "request") {
permission = entry.Request.permission;
permission = (entry as lib.RequestEntry).Request.permission;
} else if (message.type === "event") {
permission = entry.Event.permission;
permission = (entry as lib.EventEntry).Event.permission;
} else {
return;
}
@@ -205,22 +191,22 @@ class ControlConnection extends BaseConnection {
throw new Error("Should be unreachable");
}
async handleControllerConfigGetRequest() {
async handleControllerConfigGetRequest(): Promise<lib.RawConfig> {
return new lib.RawConfig(this._controller.config.serialize("control"));
}
async handleControllerConfigSetFieldRequest(request) {
async handleControllerConfigSetFieldRequest(request: lib.ControllerConfigSetFieldRequest) {
this._controller.config.set(request.field, request.value, "control");
}
async handleControllerConfigSetPropRequest(request) {
async handleControllerConfigSetPropRequest(request: lib.ControllerConfigSetPropRequest) {
let { field, prop, value } = request;
this._controller.config.setProp(field, prop, value, "control");
}
async handleHostListRequest() {
async handleHostListRequest(): Promise<lib.HostDetails[]> {
let list = [];
for (let host of this._controller.hosts.values()) {
for (let host of this._controller.hosts!.values()) {
list.push(new lib.HostDetails(
host.agent,
host.version,
@@ -233,11 +219,14 @@ class ControlConnection extends BaseConnection {
return list;
}
async handleHostSetSubscriptionsRequest(request) {
async handleHostSetSubscriptionsRequest(request: lib.HostSetSubscriptionsRequest) {
this.hostSubscriptions = { ...request };
}
hostUpdated(host, update) {
hostUpdated(
host: HostInfo,
update: lib.HostDetails
) {
if (
this.hostSubscriptions.all
|| this.hostSubscriptions.hostIds.includes(host.id)
@@ -246,15 +235,15 @@ class ControlConnection extends BaseConnection {
}
}
async handleHostGenerateTokenRequest(message) {
let hostId = message.host_id;
async handleHostGenerateTokenRequest(message: lib.HostGenerateTokenRequest) {
let hostId = message.hostId;
if (hostId === undefined) {
hostId = Math.random() * 2**31 | 0;
}
return this._controller.generateHostToken(hostId);
}
async handleHostConfigCreateRequest(request) {
async handleHostConfigCreateRequest(request: lib.HostConfigCreateRequest) {
let hostConfig = new lib.HostConfig("control");
await hostConfig.init();
@@ -272,47 +261,78 @@ class ControlConnection extends BaseConnection {
return new lib.RawConfig(hostConfig.serialize());
}
async handleInstanceDetailsGetRequest(request) {
async handleInstanceDetailsGetRequest(request: lib.InstanceDetailsGetRequest) {
let instance = this._controller.getRequestInstance(request.instanceId);
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port
if (game_port === null) {
game_port = undefined;
}
return new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host"),
instance.game_port || undefined,
assigned_host,
game_port,
instance.status,
);
}
async handleInstanceDetailsListRequest() {
let list = [];
for (let instance of this._controller.instances.values()) {
for (let instance of this._controller.instances!.values()) {
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port
if (game_port === null) {
game_port = undefined;
}
list.push(new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host") || undefined,
instance.game_port || undefined,
assigned_host,
game_port,
instance.status,
));
}
return list;
}
async handleInstanceDetailsSetSubscriptionsRequest(request) {
async handleInstanceDetailsSetSubscriptionsRequest(request: lib.InstanceDetailsSetSubscriptionsRequest) {
this.instanceSubscriptions = { ...request };
}
instanceUpdated(instance) {
instanceUpdated(instance: InstanceInfo) {
if (
this.instanceSubscriptions.all
|| this.instanceSubscriptions.instanceIds.includes(instance.id)
) {
let assigned_host: number|null|undefined = instance.config.get("instance.assigned_host");
if (assigned_host === null) {
assigned_host = undefined;
}
let game_port: number|null|undefined = instance.game_port
if (game_port === null) {
game_port = undefined;
}
this.send(new lib.InstanceDetailsUpdateEvent(
new lib.InstanceDetails(
instance.config.get("instance.name"),
instance.id,
instance.config.get("instance.assigned_host") || undefined,
instance.game_port || undefined,
assigned_host,
game_port,
instance.status,
)
));
@@ -320,22 +340,22 @@ class ControlConnection extends BaseConnection {
}
// XXX should probably add a hook for host reuqests?
async handleInstanceCreateRequest(request) {
async handleInstanceCreateRequest(request: lib.InstanceCreateRequest) {
let instanceConfig = new lib.InstanceConfig("controller");
await instanceConfig.load(request.config);
await instanceConfig.load(request.config as lib.SerializedConfig);
await this._controller.instanceCreate(instanceConfig);
}
async handleInstanceDeleteRequest(request) {
async handleInstanceDeleteRequest(request: lib.InstanceDeleteRequest) {
await this._controller.instanceDelete(request.instanceId);
}
async handleInstanceConfigGetRequest(request) {
async handleInstanceConfigGetRequest(request: lib.InstanceConfigGetRequest) {
let instance = this._controller.getRequestInstance(request.instanceId);
return new lib.InstanceConfigGetRequest.Response(instance.config.serialize("control"));
}
async handleInstanceConfigSetFieldRequest(request) {
async handleInstanceConfigSetFieldRequest(request: lib.InstanceConfigSetFieldRequest) {
let instance = this._controller.getRequestInstance(request.instanceId);
if (request.field === "instance.assigned_host") {
throw new lib.RequestError("instance.assigned_host must be set through the assign-host interface");
@@ -350,22 +370,22 @@ class ControlConnection extends BaseConnection {
await this._controller.instanceConfigUpdated(instance);
}
async handleInstanceConfigSetPropRequest(request) {
async handleInstanceConfigSetPropRequest(request: lib.InstanceConfigSetPropRequest) {
let instance = this._controller.getRequestInstance(request.instanceId);
let { field, prop, value } = request;
instance.config.setProp(field, prop, value, "control");
await this._controller.instanceConfigUpdated(instance);
}
async handleInstanceAssignRequest(request) {
async handleInstanceAssignRequest(request: lib.InstanceAssignRequest) {
await this._controller.instanceAssign(request.instanceId, request.hostId);
}
async handleInstanceSetSaveListSubscriptionsRequest(request) {
async handleInstanceSetSaveListSubscriptionsRequest(request: lib.InstanceSetSaveListSubscriptionsRequest) {
this.saveListSubscriptions = { ...request };
}
saveListUpdate(instanceId, saves) {
saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) {
if (
this.saveListSubscriptions.all
|| this.saveListSubscriptions.instanceIds.includes(instanceId)
@@ -374,7 +394,7 @@ class ControlConnection extends BaseConnection {
}
}
async handleInstanceDownloadSaveRequest(request) {
async handleInstanceDownloadSaveRequest(request: lib.InstanceDownloadSaveRequest) {
let { instanceId, name } = request;
let stream = await routes.createProxyStream(this._controller.app);
stream.filename = name;
@@ -397,7 +417,7 @@ class ControlConnection extends BaseConnection {
return stream.id;
}
async handleInstanceTransferSaveRequest(request) {
async handleInstanceTransferSaveRequest(request: lib.InstanceTransferSaveRequest) {
if (request.sourceInstanceId === request.targetInstanceId) {
throw new lib.RequestError("Source and target instance may not be the same");
}
@@ -419,12 +439,12 @@ class ControlConnection extends BaseConnection {
// Check connectivity
let sourceHostConnection = this._controller.wsServer.hostConnections.get(sourceHostId);
if (!sourceHostConnection || sourceHostConnection.closing) {
if (!sourceHostConnection || sourceHostConnection.connector.closing) {
throw new lib.RequestError("Source host is not connected to the controller");
}
let targetHostConnection = this._controller.wsServer.hostConnections.get(targetHostId);
if (!targetHostConnection || targetHostConnection.closing) {
if (!targetHostConnection || targetHostConnection.connector.closing) {
throw new lib.RequestError("Target host is not connected to the controller");
}
@@ -468,55 +488,58 @@ class ControlConnection extends BaseConnection {
}
async handleModPackListRequest() {
return [...this._controller.modPacks.values()];
return [...this._controller.modPacks!.values()];
}
async handleModPackSetSubscriptionsRequest(request) {
async handleModPackSetSubscriptionsRequest(request: lib.ModPackSetSubscriptionsRequest) {
this.modPackSubscriptions = { ...request };
}
async handleModPackCreateRequest(request) {
async handleModPackCreateRequest(request: lib.ModPackCreateRequest) {
let modPack = request.modPack;
if (this._controller.modPacks.has(modPack.id)) {
if (modPack.id === undefined) {
throw new lib.RequestError(`Mod pack need an ID to be created`);
}
if (this._controller.modPacks!.has(modPack.id)) {
throw new lib.RequestError(`Mod pack with ID ${modPack.id} already exist`);
}
this._controller.modPacks.set(modPack.id, modPack);
this._controller.modPacks!.set(modPack.id, modPack);
this._controller.modPackUpdated(modPack);
}
async handleModPackUpdateRequest(request) {
async handleModPackUpdateRequest(request: lib.ModPackUpdateRequest) {
let modPack = request.modPack;
if (!this._controller.modPacks.has(modPack.id)) {
if (modPack.id === undefined || !this._controller.modPacks!.has(modPack.id)) {
throw new lib.RequestError(`Mod pack with ID ${modPack.id} does not exist`);
}
this._controller.modPacks.set(modPack.id, modPack);
this._controller.modPacks!.set(modPack.id, modPack);
this._controller.modPackUpdated(modPack);
}
async handleModPackDeleteRequest(request) {
async handleModPackDeleteRequest(request: lib.ModPackDeleteRequest) {
let { id } = request;
let modPack = this._controller.modPacks.get(id);
let modPack = this._controller.modPacks!.get(id);
if (!modPack) {
throw new lib.RequestError(`Mod pack with ID ${id} does not exist`);
}
modPack.isDeleted = true;
this._controller.modPacks.delete(id);
this._controller.modPacks!.delete(id);
this._controller.modPackUpdated(modPack);
}
modPackUpdated(modPack) {
modPackUpdated(modPack: lib.ModPack) {
if (
this.modPackSubscriptions.all
|| this.modPackSubscriptions.modPackIds.includes(modPack.id)
|| this.modPackSubscriptions.modPackIds.includes(modPack.id!)
) {
this.send(new lib.ModPackUpdateEvent(modPack));
}
}
async handleModGetRequest(request) {
async handleModGetRequest(request: lib.ModGetRequest) {
let { name, version } = request;
let filename = `${name}_${version}.zip`;
let mod = this._controller.mods.get(filename);
let mod = this._controller.mods!.get(filename);
if (!mod) {
throw new lib.RequestError(`Mod ${filename} does not exist`);
}
@@ -524,10 +547,10 @@ class ControlConnection extends BaseConnection {
}
async handleModListRequest() {
return [...this._controller.mods.values()];
return [...this._controller.mods!.values()];
}
static termsMatchesMod(terms, mod) {
static termsMatchesMod(terms: lib.ParsedTerm[], mod: lib.ModInfo) {
for (let term of terms) {
if (term.type === "word") {
if (!lib.wordMatches(term,
@@ -537,7 +560,7 @@ class ControlConnection extends BaseConnection {
return false;
}
} else if (term.type === "attribute") {
if (!lib.wordMatches(term.value, mod[term.name])) {
if (!lib.wordMatches(term.value, mod[term.name as keyof lib.ModInfo] as string)) {
return false;
}
}
@@ -545,7 +568,7 @@ class ControlConnection extends BaseConnection {
return true;
}
async handleModSearchRequest(request) {
async handleModSearchRequest(request: lib.ModSearchRequest) {
let query = lib.parseSearchString(request.query, {
name: "word",
// version
@@ -562,8 +585,9 @@ class ControlConnection extends BaseConnection {
});
let factorioVersion = request.factorioVersion;
let results = new Map();
for (let mod of this._controller.mods.values()) {
type ModVersions = { name: string, versions: lib.ModInfo[] };
let results: Map<string, ModVersions> = new Map();
for (let mod of this._controller.mods!.values()) {
if (
mod.factorioVersion !== factorioVersion
|| !ControlConnection.termsMatchesMod(query.terms, mod)
@@ -589,14 +613,14 @@ class ControlConnection extends BaseConnection {
const sort = request.sort;
if (sort) {
const sorters = {
name: (a, b) => strcmp(a.versions[0].name, b.versions[0].name),
title: (a, b) => strcmp(a.versions[0].title, b.versions[0].title),
author: (a, b) => strcmp(a.versions[0].author, b.versions[0].author),
name: (a:ModVersions, b:ModVersions) => strcmp(a.versions[0].name, b.versions[0].name),
title: (a:ModVersions, b:ModVersions) => strcmp(a.versions[0].title, b.versions[0].title),
author: (a:ModVersions, b:ModVersions) => strcmp(a.versions[0].author, b.versions[0].author),
};
if (!Object.prototype.hasOwnProperty.call(sorters, sort)) {
throw new lib.RequestError(`Invalid value for sort: ${sort}`);
}
resultList.sort(sorters[sort]);
resultList.sort(sorters[sort as keyof typeof sorters]);
let order = request.sortOrder;
if (order === "desc") {
resultList.reverse();
@@ -615,14 +639,14 @@ class ControlConnection extends BaseConnection {
};
}
async handleModSetSubscriptionsRequest(request) {
async handleModSetSubscriptionsRequest(request: lib.ModSetSubscriptionsRequest) {
this.modSubscriptions = { ...request };
}
async handleModDownloadRequest(request) {
async handleModDownloadRequest(request: lib.ModDownloadRequest) {
let { name, version } = request;
let filename = `${name}_${version}.zip`;
let mod = this._controller.mods.get(filename);
let mod = this._controller.mods!.get(filename);
if (!mod) {
throw new lib.RequestError(`Mod ${filename} does not exist`);
}
@@ -637,16 +661,21 @@ class ControlConnection extends BaseConnection {
return stream.id;
}
async handleModDeleteRequest(request) {
async handleModDeleteRequest(request: lib.ModDeleteRequest) {
await this._controller.deleteMod(request.name, request.version);
}
async handleLogSetSubscriptionsRequest(request) {
this.logSubscriptions = { ...request };
async handleLogSetSubscriptionsRequest(request: lib.LogSetSubscriptionsRequest) {
this.logSubscriptions = {
all: request.all || false,
controller: request.controller || false,
hostIds: request.hostIds || [],
instanceIds: request.instanceIds || [],
};
this.updateLogSubscriptions();
}
modUpdated(mod) {
modUpdated(mod: lib.ModInfo) {
if (
this.modSubscriptions.all
|| this.modSubscriptions.modNames.includes(mod.name)
@@ -670,15 +699,25 @@ class ControlConnection extends BaseConnection {
}
}
async handleLogQueryRequest(request) {
async handleLogQueryRequest(request: lib.LogQueryRequest) {
let _request = {
limit: request.limit,
order: request.order,
maxLevel: request.maxLevel,
all: request.all,
controller: request.controller,
hostIds: request.hostIds,
instanceIds: request.instanceIds,
}
let observeDuration = queryLogTime.startTimer();
let { all, controller, hostIds, instanceIds } = request;
let log;
if (!all && controller && !hostIds.length && !instanceIds.length) {
log = await this._controller.queryControllerLog(request);
log = await this._controller.queryControllerLog(_request);
} else {
log = await this._controller.queryClusterLog(request);
log = await this._controller.queryClusterLog(_request);
}
observeDuration();
@@ -710,7 +749,7 @@ class ControlConnection extends BaseConnection {
return list;
}
async handleRoleCreateRequest(request) {
async handleRoleCreateRequest(request: lib.RoleCreateRequest) {
let lastId = Math.max.apply(null, [...this._controller.userManager.roles.keys()]);
// Start at 5 to leave space for future default roles
@@ -719,7 +758,7 @@ class ControlConnection extends BaseConnection {
return id;
}
async handleRoleUpdateRequest(request) {
async handleRoleUpdateRequest(request: lib.RoleUpdateRequest) {
let { id, name, description, permissions } = request;
let role = this._controller.userManager.roles.get(id);
if (!role) {
@@ -732,7 +771,7 @@ class ControlConnection extends BaseConnection {
this._controller.rolePermissionsUpdated(role);
}
async handleRoleGrantDefaultPermissionsRequest(request) {
async handleRoleGrantDefaultPermissionsRequest(request: lib.RoleGrantDefaultPermissionsRequest) {
let role = this._controller.userManager.roles.get(request.id);
if (!role) {
throw new lib.RequestError(`Role with ID ${request.id} does not exist`);
@@ -742,7 +781,7 @@ class ControlConnection extends BaseConnection {
this._controller.rolePermissionsUpdated(role);
}
async handleRoleDeleteRequest(request) {
async handleRoleDeleteRequest(request: lib.RoleDeleteRequest) {
let id = request.id;
let role = this._controller.userManager.roles.get(id);
if (!role) {
@@ -756,7 +795,7 @@ class ControlConnection extends BaseConnection {
}
}
async handleUserGetRequest(request) {
async handleUserGetRequest(request: lib.UserGetRequest) {
let name = request.name;
let user = this._controller.userManager.users.get(name);
if (!user) {
@@ -773,7 +812,7 @@ class ControlConnection extends BaseConnection {
user.banReason,
undefined,
user.playerStats,
[...user.instanceStats],
user.instanceStats,
);
}
@@ -796,11 +835,11 @@ class ControlConnection extends BaseConnection {
return list;
}
async handleUserSetSubscriptionsRequest(request) {
async handleUserSetSubscriptionsRequest(request: lib.UserSetSubscriptionsRequest) {
this.userSubscriptions = { ...request };
}
userUpdated(user) {
userUpdated(user: any) {
if (
this.userSubscriptions.all
|| this.userSubscriptions.names.includes(user.name)
@@ -816,18 +855,18 @@ class ControlConnection extends BaseConnection {
user.banReason,
user.isDeleted,
user.playerStats,
[...user.instanceStats],
user.instanceStats,
)
));
}
}
async handleUserCreateRequest(request) {
async handleUserCreateRequest(request: lib.UserCreateRequest) {
let user = this._controller.userManager.createUser(request.name);
this._controller.userUpdated(user);
}
async handleUserRevokeTokenRequest(request) {
async handleUserRevokeTokenRequest(request: lib.UserRevokeTokenRequest) {
let user = this._controller.userManager.users.get(request.name);
if (!user) {
throw new lib.RequestError(`User '${request.name}' does not exist`);
@@ -845,13 +884,13 @@ class ControlConnection extends BaseConnection {
this._controller.userUpdated(user);
}
async handleUserUpdateRolesRequest(request) {
async handleUserUpdateRolesRequest(request: lib.UserUpdateRolesRequest) {
let user = this._controller.userManager.users.get(request.name);
if (!user) {
throw new lib.RequestError(`User '${request.name}' does not exist`);
}
let resolvedRoles = new Set();
let resolvedRoles: Set<lib.Role> = new Set();
for (let roleId of request.roles) {
let role = this._controller.userManager.roles.get(roleId);
if (!role) {
@@ -866,7 +905,7 @@ class ControlConnection extends BaseConnection {
this._controller.userUpdated(user);
}
async handleUserSetAdminRequest(request) {
async handleUserSetAdminRequest(request: lib.UserSetAdminRequest) {
let { name, create, admin } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
@@ -883,7 +922,7 @@ class ControlConnection extends BaseConnection {
this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, admin));
}
async handleUserSetBannedRequest(request) {
async handleUserSetBannedRequest(request: lib.UserSetBannedRequest) {
let { name, create, banned, reason } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
@@ -901,7 +940,7 @@ class ControlConnection extends BaseConnection {
this._controller.sendTo("allInstances", new lib.InstanceBanlistUpdateEvent(name, banned, reason));
}
async handleUserSetWhitelistedRequest(request) {
async handleUserSetWhitelistedRequest(request: lib.UserSetWhitelistedRequest) {
let { name, create, whitelisted } = request;
let user = this._controller.userManager.users.get(name);
if (!user) {
@@ -918,7 +957,7 @@ class ControlConnection extends BaseConnection {
this._controller.sendTo("allInstances", new lib.InstanceWhitelistUpdateEvent(name, whitelisted));
}
async handleUserDeleteRequest(request) {
async handleUserDeleteRequest(request: lib.UserDeleteRequest) {
let name = request.name;
let user = this._controller.userManager.users.get(name);
if (!user) {
@@ -929,24 +968,24 @@ class ControlConnection extends BaseConnection {
this._controller.userManager.users.delete(name);
this._controller.userUpdated(user);
if (user.is_admin) {
if (user.isAdmin) {
this._controller.sendTo("allInstances", new lib.InstanceAdminlistUpdateEvent(name, false));
}
if (user.is_whitelisted) {
if (user.isWhitelisted) {
this._controller.sendTo("allInstances", new lib.InstanceWhitelistUpdateEvent(name, false));
}
if (user.is_banned) {
if (user.isBanned) {
this._controller.sendTo("allInstances", new lib.InstanceBanlistUpdateEvent(name, false, ""));
}
}
async handleDebugDumpWsRequest(request) {
async handleDebugDumpWsRequest(request: lib.DebugDumpWsRequest) {
this.ws_dumper = data => {
if (this.connector.connected) {
this.send(new lib.DebugWsMessageEvent(data.direction, data.content));
}
};
this.connector._socket.clusterio_ignore_dump = true;
this.connector._socket!.clusterio_ignore_dump = true;
this._controller.debugEvents.on("message", this.ws_dumper);
}
}
packages/controller/src/Controller.js → packages/controller/src/Controller.ts
+214 -204
View File
@@ -1,32 +1,39 @@
"use strict";
const compression = require("compression");
const events = require("events");
const express = require("express");
const fs = require("fs-extra");
const http = require("http");
const https = require("https");
const jwt = require("jsonwebtoken");
const path = require("path");
const stream = require("stream");
const lib = require("@clusterio/lib");
const { logger } = lib;
const HttpCloser = require("./HttpCloser");
const InstanceInfo = require("./InstanceInfo");
const metrics = require("./metrics");
const routes = require("./routes");
const UserManager = require("./UserManager");
const WsServer = require("./WsServer");
import type winston from "winston";
import type { AddressInfo } from "net";
import type { ControllerArgs } from "../controller";
import type { HostInfo } from "./HostConnection";
import type { Request, Response, Application } from "express";
const endpointDurationSummary = new lib.Summary(
import compression from "compression";
import events, { EventEmitter } from "events";
import express from "express";
import fs from "fs-extra";
import http from "http";
import https from "https";
import jwt from "jsonwebtoken";
import path from "path";
import stream from "stream";
import * as lib from "@clusterio/lib";
const { logger, Summary, Gauge } = lib;
import HttpCloser from "./HttpCloser";
import InstanceInfo, { InstanceStatus } from "./InstanceInfo";
import metrics from "./metrics";
import * as routes from "./routes";
import UserManager from "./UserManager";
import WsServer from "./WsServer";
import yargs from "yargs";
import HostConnection from "./HostConnection";
const endpointDurationSummary = new Summary(
"clusterio_controller_http_endpoint_duration_seconds",
"Time it took to respond to a an HTTP request",
{ labels: ["route"] }
);
const logSizeGauge = new lib.Gauge(
const logSizeGauge = new Gauge(
"clusterio_controller_log_bytes",
"Size of all log files currently stored on the controller."
);
@@ -35,86 +42,73 @@ const logSizeGauge = new lib.Gauge(
* Manages all controller related operations
* @alias module:controller/src/Controller
*/
class Controller {
constructor(clusterLogger, pluginInfos, configPath, config) {
export default class Controller {
clusterLogger: winston.Logger;
/** Array of plugin info objects for known plugins */
pluginInfos: lib.PluginInfo[];
configPath: string;
/** Controller config. */
config: lib.ControllerConfig;
app: Application;
/** Mapping of host id to host info */
hosts: Map<number, HostInfo> | null = null;
/** Mapping of instance id to instance info */
instances: Map<number, InstanceInfo> | null = null;
/** Mapping of mod pack id to mod pack */
modPacks: Map<number, lib.ModPack> | null = null;
/** Mapping of mod names to mod infos */
mods: Map<string, lib.ModInfo> | null = null;
/** User and roles manager for the cluster */
userManager: UserManager;
httpServer: http.Server | null = null;
httpServerCloser: HttpCloser | null = null;
httpsServer: https.Server | null = null;
httpsServerCloser: HttpCloser | null = null;
/** Mapping of plugin name to loaded plugin */
plugins: Map<string, lib.BaseControllerPlugin> = new Map();
/** WebSocket server */
wsServer: WsServer;
debugEvents: events.EventEmitter = new events.EventEmitter();
private _events: events.EventEmitter = new events.EventEmitter();
// Possible states are new, starting, running, stopping, stopped
private _state: string = "new";
private _shouldStop: boolean = false;
_fallbackedRequests: Map<lib.RequestClass<unknown, unknown>, lib.RequestHandler<unknown, unknown>> = new Map();
_registeredRequests: Map<lib.RequestClass<unknown, unknown>, lib.RequestHandler<unknown, unknown>> = new Map();
_registeredEvents = new Map();
_snoopedEvents = new Map();
devMiddleware: any | null = null;
logDirectory: string = "";
clusterLogIndex: lib.LogIndex | null = null;
clusterLogBuildInterval: NodeJS.Timer | null = null;
constructor(
clusterLogger: winston.Logger,
pluginInfos: any[],
configPath: string,
config: lib.ControllerConfig
) {
this.clusterLogger = clusterLogger;
/**
* Array of plugin info objects for known plugins
* @type {Array<object>}
*/
this.pluginInfos = pluginInfos;
this.configPath = configPath;
/**
* Controller config.
* @type {module:lib.ControllerConfig}
*/
this.config = config;
this.app = express();
this.app.locals.controller = this;
this.app.locals.streams = new Map();
/**
* Mapping of host id to host info
* @type {Map<number, Object>}
*/
this.hosts = null;
/**
* Mapping of instance id to instance info
* @type {Map<number, module:controller/src/InstanceInfo>}
*/
this.instances = null;
/**
* Mapping of mod pack id to mod pack
* @type {Map<number, module:lib.ModPack>}
*/
this.modPacks = null;
/**
* Mapping of mod names to mod infos
* @type {Map<string, module:lib.ModInfo>}
*/
this.mods = null;
/**
* User and roles manager for the cluster
* @type {module:controller/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<string, module:lib.BaseControllerPlugin>}
*/
this.plugins = new Map();
/**
* WebSocket server
* @type {module:controller/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._fallbackedRequests = new Map();
this._registeredRequests = new Map();
this._registeredEvents = new Map();
this._snoopedEvents = new Map();
this.devMiddleware = null;
this.userManager = new UserManager(this.config);
}
async start(args) {
async start(args: ControllerArgs) {
if (this._state !== "new") {
throw new Error(`Cannot start in state ${this._state}`);
}
@@ -148,13 +142,16 @@ class Controller {
])).reduce((a, v) => a + v, 0);
}
async _startInternal(args) {
async _startInternal(args: ControllerArgs) {
this.logDirectory = args.logDirectory;
this.clusterLogIndex = await lib.LogIndex.load(path.join(this.logDirectory, "cluster"));
this.clusterLogBuildInterval = setInterval(() => {
this.clusterLogIndex.buildIndex().catch(
err => logger.error(`Error building cluster log index:\n${err.stack}`)
);
if (this.clusterLogIndex) {
this.clusterLogIndex.buildIndex().catch(
err => logger.error(`Error building cluster log index:\n${err.stack}`)
);
}
}, 600e3);
logSizeGauge.callback = async () => { logSizeGauge.set(await this.logSize()); };
@@ -165,6 +162,7 @@ class Controller {
const webpack = require("webpack");
const webpackDevMiddleware = require("webpack-dev-middleware");
const webpackConfigs = [];
if (args.dev) {
webpackConfigs.push(require("../webpack.config")({}));
}
@@ -194,7 +192,6 @@ class Controller {
this.hosts = await Controller.loadHosts(path.join(databaseDirectory, "hosts.json"));
this.instances = await Controller.loadInstances(path.join(databaseDirectory, "instances.json"));
this.modPacks = await Controller.loadModPacks(path.join(databaseDirectory, "mod-packs.json"));
this.userManager = new UserManager(this.config);
await this.userManager.load(path.join(databaseDirectory, "users.json"));
let modsDirectory = this.config.get("controller.mods_directory");
@@ -204,7 +201,7 @@ class Controller {
this.config.on("fieldChanged", (group, field, prev) => {
lib.invokeHook(this.plugins, "onControllerConfigFieldChanged", group, field, prev);
});
for (let instance of this.instances.values()) {
for (let instance of this.instances!.values()) {
this.addInstanceHooks(instance);
}
@@ -230,7 +227,9 @@ class Controller {
Controller.addAppRoutes(this.app, this.pluginInfos);
if (!args.dev) {
let manifest = await Controller.loadJsonObject(path.join(__dirname, "..", "dist", "web", "manifest.json"));
let manifestPath = path.join(__dirname, "..", "..", "web", "manifest.json")
let manifest = await Controller.loadJsonObject(manifestPath);
if (!manifest["main.js"]) {
logger.error("Missing main.js entry in dist/web/manifest.json");
}
@@ -247,16 +246,16 @@ class Controller {
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}`);
logger.info(`Listening for HTTP on port ${(this.httpServer.address() as AddressInfo).port}`);
}
if (httpsPort) {
if (httpsPort && tls_cert && tls_key) {
let certificate, privateKey;
try {
certificate = await fs.readFile(tls_cert);
privateKey = await fs.readFile(tls_key);
} catch (err) {
} catch (err: any) {
throw new lib.StartupError(
`Error loading ssl certificate: ${err.message}`
);
@@ -268,7 +267,7 @@ class Controller {
}, 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(`Listening for HTTPS on port ${(this.httpsServer.address() as AddressInfo).port}`);
}
logger.info("Started controller");
@@ -299,7 +298,11 @@ class Controller {
// This function should never throw.
this._state = "stopping";
logger.info("Stopping controller");
clearInterval(this.clusterLogBuildInterval);
if (this.clusterLogBuildInterval) {
clearInterval(this.clusterLogBuildInterval);
}
if (this.clusterLogIndex) {
await this.clusterLogIndex.save();
}
@@ -336,23 +339,26 @@ class Controller {
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()); }
if (this.httpServer && this.httpServer.listening && this.httpServerCloser) {
stopTasks.push(this.httpServerCloser.close());
}
if (this.httpsServer && this.httpsServer.listening && this.httpsServerCloser) {
stopTasks.push(this.httpsServerCloser.close());
}
await Promise.all(stopTasks);
logger.info("Goodbye");
}
static async loadHosts(filePath) {
let serialized;
static async loadHosts(filePath: string): Promise<Map<number, HostInfo>> {
let serialized: [number, HostInfo][];
try {
serialized = JSON.parse(await fs.readFile(filePath));
serialized = JSON.parse(await fs.readFile(filePath, { encoding: 'utf8' }));
} catch (err) {
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
return new Map();
}
@@ -364,60 +370,64 @@ class Controller {
return new Map(serialized);
}
static async saveHosts(filePath, hosts) {
static async saveHosts(filePath: string, hosts: Map<number, HostInfo>) {
await lib.safeOutputFile(filePath, JSON.stringify([...hosts.entries()], null, 4));
}
static async loadInstances(filePath) {
static async loadInstances(filePath: string): Promise<Map<number, InstanceInfo>> {
logger.info(`Loading ${filePath}`);
let instances = new Map();
try {
let serialized = JSON.parse(await fs.readFile(filePath));
let serialized = JSON.parse(await fs.readFile(filePath, { encoding: "utf8" }));
for (let serializedConfig of serialized) {
let instanceConfig = new lib.InstanceConfig("controller");
await instanceConfig.load(serializedConfig);
let status = instanceConfig.get("instance.assigned_host") === null ? "unassigned" : "unknown";
let instance = new InstanceInfo({ config: instanceConfig, status });
let instance = new InstanceInfo({
config: instanceConfig,
status: status as InstanceStatus,
});
instances.set(instanceConfig.get("instance.id"), instance);
}
} catch (err) {
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
return new Map();
}
return instances;
}
static async saveInstances(filePath, instances) {
static async saveInstances(filePath: string, instances: Map<number, InstanceInfo>) {
let serialized = [];
for (let instance of instances.values()) {
for (const instance of instances.values()) {
serialized.push(instance.config.serialize());
}
await lib.safeOutputFile(filePath, JSON.stringify(serialized, null, 4));
}
static async loadModPacks(filePath) {
static async loadModPacks(filePath: string): Promise<Map<number, lib.ModPack>> {
let json;
try {
json = JSON.parse(await fs.readFile(filePath));
} catch (err) {
json = JSON.parse(await fs.readFile(filePath, { encoding : "utf8" }));
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
return new Map();
}
return new Map(json.map(e => [e.id, lib.ModPack.fromJSON(e)]));
return new Map(json.map((e: any) => [e.id, lib.ModPack.fromJSON(e)]));
}
static async saveModPacks(filePath, modPacks) {
static async saveModPacks(filePath: string, modPacks: Map<number, lib.ModPack>) {
await lib.safeOutputFile(filePath, JSON.stringify([...modPacks.values()], null, 4));
}
static async loadModInfos(modsDirectory) {
static async loadModInfos(modsDirectory: string): Promise<Map<string,lib.ModInfo>> {
let mods = new Map();
for (let entry of await fs.readdir(modsDirectory, { withFileTypes: true })) {
if (entry.isDirectory()) {
@@ -433,7 +443,7 @@ class Controller {
let modInfo;
try {
modInfo = await lib.ModInfo.fromModFile(path.join(modsDirectory, entry.name));
} catch (err) {
} catch (err: any) {
logger.error(`Error loading mod ${entry.name}: ${err.message}`);
continue;
}
@@ -451,11 +461,11 @@ class Controller {
return mods;
}
static async loadJsonObject(filePath, throwOnMissing = false) {
static async loadJsonObject(filePath: string, throwOnMissing: boolean = false): Promise<any> {
let manifest = {};
try {
manifest = JSON.parse(await fs.readFile(filePath));
} catch (err) {
manifest = JSON.parse(await fs.readFile(filePath, { encoding: 'utf8' }));
} catch (err: any) {
if (!throwOnMissing && err.code !== "ENOENT") {
throw err;
}
@@ -463,16 +473,16 @@ class Controller {
return manifest;
}
async deleteMod(name, version) {
async deleteMod(name: string, version: string): Promise<void> {
let filename = `${name}_${version}.zip`;
let mod = this.mods.get(filename);
let mod = this.mods!.get(filename);
if (!mod) {
throw new Error(`Mod ${filename} does not exist`);
}
let modsDirectory = this.config.get("controller.mods_directory");
await fs.unlink(path.join(modsDirectory, filename));
this.mods.delete(filename);
this.mods!.delete(filename);
mod.isDeleted = true;
this.modUpdated(mod);
}
@@ -480,32 +490,32 @@ class Controller {
/**
* Query controller log
*
* @param {module:lib~QueryLogFilter} filter -
* @param filter -
* Filter to limit entries with. Note that only the controller log can
* be queried from this function.
* @returns {Promise<Array<Object>>} log entries matching the filter
* @returns log entries matching the filter
*/
async queryControllerLog(filter) {
return await lib.queryLog(
path.join(this.logDirectory, "controller"), filter,
async queryControllerLog(filter: lib.QueryLogFilter): Promise<object[]> {
return lib.queryLog(
path.join(this.logDirectory, "controller"), filter, null,
);
}
/**
* Query cluster log
*
* @param {module:lib~QueryLogFilter} filter -
* @param filter -
* Filter to limit entries with.
* @returns {Promise<Array<Object>>} log entries matching the filter
* @returns log entries matching the filter
*/
async queryClusterLog(filter) {
async queryClusterLog(filter: lib.QueryLogFilter): Promise<object[]> {
return await lib.queryLog(
path.join(this.logDirectory, "cluster"), filter, this.clusterLogIndex,
);
}
static addAppRoutes(app, pluginInfos) {
app.use((req, res, next) => {
static addAppRoutes(app: Application, pluginInfos: any[]) {
app.use((req: Request, res: Response, next) => {
let start = process.hrtime.bigint();
stream.finished(res, () => {
let routePath = "static";
@@ -521,7 +531,7 @@ class Controller {
// Set folder to serve static content from (the website)
const staticOptions = { immutable: true, maxAge: 1000 * 86400 * 365 };
app.use("/static", express.static(path.join(__dirname, "..", "dist", "web", "static"), staticOptions));
app.use("/static", express.static(path.join(__dirname, "..", "..", "..", "dist", "web", "static"), staticOptions));
app.use("/static", express.static("static", staticOptions)); // Used for data export files
// Add API routes
@@ -545,17 +555,17 @@ class Controller {
/**
* Generate access token for host
*
* @param {number} hostId - ID of host to generate a token for.
* @returns {string} access token for host
* @returns access token for host
* @param hostId - ID of host to generate a token for.
*/
generateHostToken(hostId) {
generateHostToken(hostId: number): string {
return jwt.sign(
{ aud: "host", host: hostId },
Buffer.from(this.config.get("controller.auth_secret"), "base64")
);
}
hostUpdated(host) {
hostUpdated(host: HostInfo) {
let update = new lib.HostDetails(
host.agent,
host.version,
@@ -577,12 +587,11 @@ class Controller {
/**
* Get instance by ID for a request
*
* @param {number} instanceId - ID of instance to get.
* @returns {object} instance
* @param instanceId - ID of instance to get.
* @throws {module:lib.RequestError} if the instance does not exist.
*/
getRequestInstance(instanceId) {
let instance = this.instances.get(instanceId);
getRequestInstance(instanceId:number):InstanceInfo {
let instance = this.instances!.get(instanceId);
if (!instance) {
throw new lib.RequestError(`Instance with ID ${instanceId} does not exist`);
}
@@ -602,13 +611,13 @@ class Controller {
* let instance = await controller.instanceAssign(instanceConfig);
* await controller.instanceAssign(instance.id, hostId);
*
* @param {module:lib.InstanceConfig} instanceConfig -
* @param instanceConfig -
* Config to base newly created instance on.
* @returns {module:controller/src/InstanceInfo} The created instance
* @returns The created instance
*/
async instanceCreate(instanceConfig) {
async instanceCreate(instanceConfig: lib.InstanceConfig): Promise<InstanceInfo> {
let instanceId = instanceConfig.get("instance.id");
if (this.instances.has(instanceId)) {
if (this.instances!.has(instanceId)) {
throw new lib.RequestError(`Instance with ID ${instanceId} already exists`);
}
@@ -639,9 +648,10 @@ class Controller {
instanceConfig.set("factorio.settings", settings);
let instance = new InstanceInfo({ config: instanceConfig, status: "unassigned" });
this.instances.set(instanceId, instance);
this.instances!.set(instanceId, instance);
await lib.invokeHook(this.plugins, "onInstanceStatusChanged", instance, null);
this.addInstanceHooks(instance);
return instance;
}
/**
@@ -654,15 +664,15 @@ class Controller {
* Note: this will not transfer any files or saves stored on the host
* the instance was previously assgined to the new one.
*
* @param {number} instanceId - ID of Instance to assign.
* @param {number} hostId - ID of host to assign instance to.
* @param instanceId - ID of Instance to assign.
* @param hostId - ID of host to assign instance to.
*/
async instanceAssign(instanceId, hostId) {
async instanceAssign(instanceId: number, hostId?: number) {
let instance = this.getRequestInstance(instanceId);
// Check if target host is connected
let newHostConnection;
if (hostId !== null) {
let newHostConnection: HostConnection | undefined;
if (hostId !== undefined) {
newHostConnection = this.wsServer.hostConnections.get(hostId);
if (!newHostConnection) {
// The case of the host not getting the assign instance message
@@ -685,7 +695,7 @@ class Controller {
// Assign to target
instance.config.set("instance.assigned_host", hostId);
if (hostId !== null) {
if (hostId !== null && newHostConnection) {
await newHostConnection.send(
new lib.InstanceAssignInternalRequest(instanceId, instance.config.serialize("host"))
);
@@ -701,15 +711,15 @@ class Controller {
* Permanently deletes the instance from its assigned host (if assigned)
* and the controller. This action cannot be undone.
*
* @param {number} instanceId - ID of instance to delete.
* @param instanceId - ID of instance to delete.
*/
async instanceDelete(instanceId) {
async instanceDelete(instanceId: number) {
let instance = this.getRequestInstance(instanceId);
let hostId = instance.config.get("instance.assigned_host");
if (hostId !== null) {
await this.sendTo({ hostId }, new lib.InstanceDeleteInternalRequest(instanceId));
}
this.instances.delete(instanceId);
this.instances!.delete(instanceId);
let prev = instance.status;
instance.status = "deleted";
@@ -723,10 +733,10 @@ class Controller {
* Used to push config changes to the assigned host when the config of
* an instance has changed.
*
* @param {module:controller/src/InstanceInfo} instance -
* @param instance -
* Instance to nodify the config updated.
*/
async instanceConfigUpdated(instance) {
async instanceConfigUpdated(instance: InstanceInfo) {
let hostId = instance.config.get("instance.assigned_host");
if (hostId !== null) {
let connection = this.wsServer.hostConnections.get(hostId);
@@ -738,8 +748,8 @@ class Controller {
}
}
addInstanceHooks(instance) {
instance.config.on("fieldChanged", (group, field, prev) => {
addInstanceHooks(instance: InstanceInfo) {
instance.config.on("fieldChanged", (group: lib.ConfigGroup, field: string, prev: any) => {
if (group.name === "instance" && field === "name") {
this.instanceUpdated(instance);
}
@@ -750,7 +760,7 @@ class Controller {
this.instanceUpdated(instance);
}
instanceUpdated(instance) {
instanceUpdated(instance: InstanceInfo) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
@@ -760,7 +770,7 @@ class Controller {
}
}
saveListUpdate(instanceId, saves) {
saveListUpdate(instanceId: number, saves: lib.SaveDetails[]) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
@@ -770,7 +780,7 @@ class Controller {
}
}
modPackUpdated(modPack) {
modPackUpdated(modPack: lib.ModPack) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
@@ -781,7 +791,7 @@ class Controller {
lib.invokeHook(this.plugins, "onModPackUpdated", modPack);
}
modUpdated(mod) {
modUpdated(mod: lib.ModInfo) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
@@ -793,7 +803,7 @@ class Controller {
lib.invokeHook(this.plugins, "onModUpdated", mod);
}
userUpdated(user) {
userUpdated(user: lib.User) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.connector.closing) {
continue;
@@ -806,9 +816,9 @@ class Controller {
/**
* Notify connected control clients under the given user that the
* permissions for this user may have changed.
* @param {module:lib.User} user - User permisions updated for.
* @param user - User permisions updated for.
*/
userPermissionsUpdated(user) {
userPermissionsUpdated(user: lib.User) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.user === user) {
controlConnection.send(
@@ -825,9 +835,9 @@ class Controller {
/**
* Notify connected control clients with the given role that the
* permissions may have changed.
* @param {module:lib.Role} role - Role permisions updated for.
* @param role - Role permisions updated for.
*/
rolePermissionsUpdated(role) {
rolePermissionsUpdated(role: lib.Role) {
for (let controlConnection of this.wsServer.controlConnections.values()) {
if (controlConnection.user.roles.has(role)) {
controlConnection.send(
@@ -862,11 +872,12 @@ class Controller {
ControllerPluginClass = await lib.loadControllerPluginClass(pluginInfo);
}
//@ts-ignore //TODO::: metrics is lib.Counter but ControllerPluginClass expect any[], can't figure out how the metrics is used.
let controllerPlugin = new ControllerPluginClass(pluginInfo, this, metrics, logger);
await controllerPlugin.init();
this.plugins.set(pluginInfo.name, controllerPlugin);
} catch (err) {
} catch (err: any) {
throw new lib.PluginError(pluginInfo.name, err);
}
@@ -883,14 +894,14 @@ class Controller {
* @param {*} args - Arguments to the .listen() call on the server.
* @returns {Promise} promise that resolves the server is listening.
*/
listen(server, ...args) {
listen(server: http.Server|https.Server, ...args: any[]): Promise<void> {
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) {
function wrapError(err: any) {
reject(new lib.StartupError(
`Server listening failed: ${err.message}`
));
@@ -908,13 +919,13 @@ class Controller {
/**
* Returns the URL needed to connect to the controller.
*
* @returns {string} controller URL.
* @returns controller URL.
*/
getControllerUrl() {
getControllerUrl(): string {
return Controller.calculateControllerUrl(this.config);
}
static calculateControllerUrl(config) {
static calculateControllerUrl(config: lib.ControllerConfig) {
let url = config.get("controller.external_address");
if (!url) {
if (config.get("controller.https_port")) {
@@ -932,24 +943,25 @@ class Controller {
* @param {string} route - route the interface is served under.
* @returns {function} Experess.js route handler.
*/
static serveWeb(route) {
static serveWeb(route: string) {
// 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 routeDepth = (route.match(/\//g) || []).length - 1 - Number(route.slice(-1) === "/");
return function(req: Request, res: Response, next: Function) {
let depth = routeDepth + Number(req.path.slice(-1) === "/");
let webRoot = "../".repeat(depth) || "./";
let staticRoot = webRoot;
let mainBundle;
let mainBundle: string = "";
if (res.app.locals.mainBundle) {
mainBundle = res.app.locals.mainBundle;
} else {
let stats = res.locals.webpack.devMiddleware.stats.stats[0];
mainBundle = stats.toJson().assetsByChunkName["main"];
}
fs.readFile(path.join(__dirname, "..", "web", "index.html"), "utf8").then((content) => {
fs.readFile(path.join(__dirname, "..", "..", "..", "web", "index.html"), "utf8").then((content) => {
res.type("text/html");
res.send(content
.replace(/__CLUSTER_NAME__/g, res.app.locals.controller.config.get("controller.name"))
@@ -963,7 +975,7 @@ class Controller {
};
}
handle(Class, handler) {
handle(Class: any, handler: any) {
if (Class.type === "request") {
this.handleRequest(Class, handler);
} else if (Class.type === "event") {
@@ -973,7 +985,7 @@ class Controller {
}
}
handleRequest(Request, handler) {
handleRequest(Request: any, handler: any) {
if (!lib.Link._requestsByClass.has(Request)) {
throw new Error(`Unregistered Request class ${Request.name}`);
}
@@ -983,7 +995,7 @@ class Controller {
this._registeredRequests.set(Request, handler);
}
fallbackRequest(Request, handler) {
fallbackRequest(Request: any, handler: any) {
if (!lib.Link._requestsByClass.has(Request)) {
throw new Error(`Unregistered Request class ${Request.name}`);
}
@@ -993,7 +1005,7 @@ class Controller {
this._fallbackedRequests.set(Request, handler);
}
handleEvent(Event, handler) {
handleEvent(Event: any, handler: any) {
if (!lib.Link._eventsByClass.has(Event)) {
throw new Error(`Unregistered Event class ${Event.name}`);
}
@@ -1003,7 +1015,7 @@ class Controller {
this._registeredEvents.set(Event, handler);
}
snoopEvent(Event, handler) {
snoopEvent(Event: any, handler: any) {
if (!lib.Link._eventsByClass.has(Event)) {
throw new Error(`Unregistered Event class ${Event.name}`);
}
@@ -1020,13 +1032,13 @@ class Controller {
* destination argument supports address shorthands, see {@link
* module:lib.Address.fromShorthand}
*
* @param {*|module:lib.Address} address - Where to send it.
* @param {*} requestOrEvent - The request or event to send.
* @returns {Promise<*>|undefined}
* @param address - Where to send it.
* @param requestOrEvent - The request or event to send.
* @returns
* Promise that resolves to the response if a request was sent or
* undefined if it was an event.
*/
sendTo(address, requestOrEvent) {
sendTo(address: lib.AddressShorthand, requestOrEvent: any): Promise<any> | void {
let dst = lib.Address.fromShorthand(address);
if (requestOrEvent.constructor.type === "request") {
return this.sendRequest(requestOrEvent, dst);
@@ -1037,7 +1049,7 @@ class Controller {
throw Error(`Unknown type ${requestOrEvent.constructor.type}.`);
}
async sendRequest(request, dst) {
async sendRequest(request: any, dst: lib.Address) {
let connection;
if (dst.type === lib.Address.controller) {
throw new Error("controller as dst is not supported.");
@@ -1072,7 +1084,7 @@ class Controller {
return connection.sendRequest(request, dst);
}
sendEvent(event, dst) {
sendEvent<T>(event: lib.Event<T>, dst: lib.Address) {
let connection;
if (dst.type === lib.Address.controller) {
throw new Error("controller as dst is not supported.");
@@ -1081,7 +1093,7 @@ class Controller {
connection = this.wsServer.controlConnections.get(dst.id);
} else if (dst.type === lib.Address.instance) {
let instance = this.instances.get(dst.id);
let instance = this.instances!.get(dst.id);
if (!instance) {
return;
}
@@ -1119,7 +1131,7 @@ class Controller {
}
}
sendToHostByInstanceId(requestOrEvent) {
async sendToHostByInstanceId(requestOrEvent: any) {
if (requestOrEvent.constructor.type === "request") {
return this.sendRequestToHostByInstanceId(requestOrEvent);
}
@@ -1129,7 +1141,7 @@ class Controller {
throw Error(`Unknown type ${requestOrEvent.constructor.type}.`);
}
async sendRequestToHostByInstanceId(request) {
async sendRequestToHostByInstanceId(request: any) {
let instance = this.getRequestInstance(request.instanceId);
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
@@ -1142,7 +1154,7 @@ class Controller {
return await connection.send(request);
}
sendEventToHostByInstanceId(event) {
sendEventToHostByInstanceId(event: any) {
let instance = this.getRequestInstance(event.instanceId);
let hostId = instance.config.get("instance.assigned_host");
if (hostId === null) {
@@ -1155,5 +1167,3 @@ class Controller {
connection.send(event);
}
}
module.exports = Controller;
packages/controller/src/ControllerRouter.js → packages/controller/src/ControllerRouter.ts
+34 -31
View File
@@ -1,39 +1,36 @@
"use strict";
const lib = require("@clusterio/lib");
import type Controller from "./Controller";
import type BaseConnection from "./BaseConnection";
class ControllerRouter {
/** @type {module:controller/src/Controller} */
controller;
import * as lib from "@clusterio/lib";
import HostConnection from "./HostConnection";
import ControlConnection from "./ControlConnection";
const { logger } = lib;
pendingRequests = new Map();
export default class ControllerRouter {
pendingRequests: Map<any, any> = new Map();
constructor(controller) {
this.controller = controller;
}
addHostConnection(connection) {
this.hostConnections.set(connection.id, connection);
}
addControlConnection(connection) {
this.controlConnections.set(connection.id, connection);
}
constructor(
public controller: Controller
) {}
/**
* @param {module:lib.Link} origin - Source link of the message.
* @param {module:lib.Message} message
* Link the message originated from.
* @param {boolean} hasFallback - true if fallback handling is available.
* @returns {boolean} true if the message was handled
* @param origin - Source link of the message.
* @param message - Link the message originated from.
* @param hasFallback - true if fallback handling is available.
* @returns true if the message was handled
*/
forwardMessage(origin, message, hasFallback) {
forwardMessage(
origin: ControlConnection | HostConnection,
message: lib.MessageSrcDst,
hasFallback: boolean
): boolean {
if (!["request", "response", "responseError", "event"].includes(message.type)) {
throw new Error(`Message type ${message.type} can't be forwarded`);
}
let dst = message.dst;
let nextHop;
let msg;
let dst: lib.Address = message.dst;
let nextHop: ControlConnection | HostConnection | undefined;
let msg: string = "";
if (dst.type === lib.Address.broadcast) {
this.broadcastMessage(origin, message);
return true;
@@ -43,7 +40,7 @@ class ControllerRouter {
msg = `Host ${dst.id} is offline`;
}
} else if (dst.type === lib.Address.instance) {
let instance = this.controller.instances.get(dst.id);
let instance = this.controller.instances!.get(dst.id);
if (!instance) {
msg = `Instance ${dst.id} does not exist`;
} else {
@@ -67,7 +64,7 @@ class ControllerRouter {
}
if (nextHop === origin) {
msg = `Message would return back to sender ${origin.dst}.`;
msg = `Message would return back to sender ${origin.connector.dst}.`;
nextHop = undefined;
}
@@ -85,7 +82,7 @@ class ControllerRouter {
if (nextHop) {
if (message.type === "request") {
nextHop.forwardRequest(message, origin);
nextHop.forwardRequest(message as lib.MessageRequest, origin);
} else {
nextHop.connector.send(message);
}
@@ -96,7 +93,10 @@ class ControllerRouter {
return true;
}
broadcastMessage(origin, message) {
broadcastMessage(
origin: lib.Link,
message: lib.MessageSrcDst,
) {
let dst = message.dst;
if (message.type !== "event") {
this.warnUnrouted(message, `Unexpected broadcast of ${message.type}`);
@@ -117,7 +117,10 @@ class ControllerRouter {
}
}
warnUnrouted(message, msg) {
warnUnrouted(
message: lib.MessageSrcDst,
msg: string
) {
let dst = message.dst;
let baseMsg = `No destination for ${message.constructor.name} routed from ${message.src} to ${dst}`;
logger.warn(msg ? `${baseMsg}: ${msg}.` : `${baseMsg}.`);
packages/controller/src/HostConnection.js → packages/controller/src/HostConnection.ts
+60 -35
View File
@@ -1,19 +1,41 @@
"use strict";
const lib = require("@clusterio/lib");
import type Controller from "./Controller";
import type WsServerConnector from "./WsServerConnector";
import * as lib from "@clusterio/lib";
const { logger, PlayerStats } = lib;
const BaseConnection = require("./BaseConnection");
const InstanceInfo = require("./InstanceInfo");
import BaseConnection from "./BaseConnection";
import InstanceInfo, { InstanceStatus } from "./InstanceInfo";
export interface HostInfo {
agent: string;
id: number;
name: string;
version: string;
plugins: [number, any][];
public_address: string | undefined;
}
/**
* Represents the connection to a host
*
* @extends module:controller/src/BaseConnection
* @alias module:controller/src/HostConnection
*/
class HostConnection extends BaseConnection {
constructor(registerData, connector, controller) {
export default class HostConnection extends BaseConnection {
declare connector: WsServerConnector;
private _agent: any;
private _id: any;
private _name: any;
private _version: any;
plugins: Map<string, Object>;
constructor(
registerData: any,
connector: WsServerConnector,
controller: Controller,
) {
super(connector, controller);
this._agent = registerData.agent;
@@ -23,7 +45,7 @@ class HostConnection extends BaseConnection {
this.plugins = new Map(Object.entries(registerData.plugins));
this._checkPluginVersions();
this._controller.hosts.set(this._id, {
this._controller.hosts!.set(this._id, {
agent: this._agent,
id: this._id,
name: this._name,
@@ -36,14 +58,14 @@ class HostConnection extends BaseConnection {
// eslint-disable-next-line no-loop-func
this.connector.on(event, () => {
for (let plugin of this._controller.plugins.values()) {
plugin.onHostConnectionEvent(this, event);
plugin.onHostConnectionEvent(this, event as "connect"|"drop"|"resume"|"close");
}
});
}
this.connector.on("close", () => {
// Update status to unknown for instances on this host.
for (let instance of this._controller.instances.values()) {
for (let instance of this._controller.instances!.values()) {
if (instance.config.get("instance.assigned_host") !== this._id) {
continue;
}
@@ -62,7 +84,7 @@ class HostConnection extends BaseConnection {
this.handle(lib.InstancePlayerUpdateEvent, this.handleInstancePlayerUpdateEvent.bind(this));
}
validateIngress(message) {
validateIngress(message: lib.MessageSrcDst) {
let origin = this.connector.dst;
switch (message.src.type) {
case lib.Address.control:
@@ -78,7 +100,7 @@ class HostConnection extends BaseConnection {
break;
case lib.Address.instance:
let instance = this._controller.instances.get(message.src.id);
let instance = this._controller.instances!.get(message.src.id);
if (!instance || instance.config.get("instance.assigned_host") !== this._id) {
throw new lib.InvalidMessage(
`Received message with invalid src ${message.src} from ${origin}`
@@ -129,8 +151,8 @@ class HostConnection extends BaseConnection {
return this._id;
}
async handleInstanceStatusChangedEvent(request) {
let instance = this._controller.instances.get(request.instanceId);
async handleInstanceStatusChangedEvent(request: lib.InstanceStatusChangedEvent) {
let instance = this._controller.instances!.get(request.instanceId);
// It's possible to get updates from an instance that does not exist
// or is not assigned to the host it originated from if it was
@@ -151,16 +173,16 @@ class HostConnection extends BaseConnection {
}
let prev = instance.status;
instance.status = request.status;
instance.game_port = request.gamePort;
instance.status = request.status as lib.InstanceStatus;
instance.game_port = request.gamePort || null;
logger.verbose(`Instance ${instance.config.get("instance.name")} State: ${instance.status}`);
this._controller.instanceUpdated(instance);
await lib.invokeHook(this._controller.plugins, "onInstanceStatusChanged", instance, prev);
}
async handleInstancesUpdateRequest(request) {
async handleInstancesUpdateRequest(request: lib.InstancesUpdateRequest) {
// Push updated instance configs
for (let instance of this._controller.instances.values()) {
for (let instance of this._controller.instances!.values()) {
if (instance.config.get("instance.assigned_host") === this._id) {
await this.send(
new lib.InstanceAssignInternalRequest(instance.id, instance.config.serialize("host"))
@@ -171,9 +193,9 @@ class HostConnection extends BaseConnection {
// Assign instances the host has but controller does not
for (let instanceData of request.instances) {
let instanceConfig = new lib.InstanceConfig("controller");
await instanceConfig.load(instanceData.config, "host");
await instanceConfig.load(instanceData.config as lib.SerializedConfig, "host");
let controllerInstance = this._controller.instances.get(instanceConfig.get("instance.id"));
let controllerInstance = this._controller.instances!.get(instanceConfig.get("instance.id"));
if (controllerInstance) {
// Check if this instance is assigned somewhere else.
if (controllerInstance.config.get("instance.assigned_host") !== this._id) {
@@ -186,7 +208,7 @@ class HostConnection extends BaseConnection {
// Already have this instance, update state instead
if (controllerInstance.status !== instanceData.status) {
let prev = controllerInstance.status;
controllerInstance.status = instanceData.status;
controllerInstance.status = instanceData.status as InstanceStatus;
logger.verbose(`Instance ${instanceConfig.get("instance.name")} State: ${instanceData.status}`);
this._controller.instanceUpdated(controllerInstance);
await lib.invokeHook(
@@ -197,8 +219,8 @@ class HostConnection extends BaseConnection {
}
instanceConfig.set("instance.assigned_host", this._id);
let newInstance = new InstanceInfo({ config: instanceConfig, status: instanceData.status });
this._controller.instances.set(instanceConfig.get("instance.id"), newInstance);
let newInstance = new InstanceInfo({ config: instanceConfig, status: instanceData.status as InstanceStatus });
this._controller.instances!.set(instanceConfig.get("instance.id"), newInstance);
this._controller.addInstanceHooks(newInstance);
await this.send(
new lib.InstanceAssignInternalRequest(
@@ -209,30 +231,30 @@ class HostConnection extends BaseConnection {
}
// Push lists to make sure they are in sync.
let adminlist = [];
let banlist = [];
let whitelist = [];
let adminlist: Set<string> = new Set();
let banlist: Map<string, string> = new Map();
let whitelist: Set<string> = new Set();
for (let user of this._controller.userManager.users.values()) {
if (user.isAdmin) {
adminlist.push(user.name);
adminlist.add(user.name);
}
if (user.isBanned) {
banlist.push([user.name, user.banReason]);
banlist.set(user.name, user.banReason);
}
if (user.isWhitelisted) {
whitelist.push(user.name);
whitelist.add(user.name);
}
}
await this.send(new lib.SyncUserListsEvent(adminlist, banlist, whitelist));
}
async handleInstanceSaveListUpdateEvent(event) {
async handleInstanceSaveListUpdateEvent(event: lib.InstanceSaveListUpdateEvent) {
this._controller.saveListUpdate(event.instanceId, event.saves);
}
async handleLogMessageEvent(event) {
async handleLogMessageEvent(event: lib.LogMessageEvent) {
this._controller.clusterLogger.log({
...event.info,
host_id: this._id,
@@ -240,7 +262,8 @@ class HostConnection extends BaseConnection {
});
}
async handleInstancePlayerUpdateEvent(event, src) {
async handleInstancePlayerUpdateEvent(event: lib.InstancePlayerUpdateEvent, src: lib.Address) {
let instanceId = src.id;
let user = this._controller.userManager.users.get(event.name);
if (!user) {
@@ -252,11 +275,15 @@ class HostConnection extends BaseConnection {
} else if (event.type === "leave") {
user.notifyLeave(instanceId);
}
user.instanceStats.set(instanceId, new PlayerStats(event.stats));
if (event.stats) {
user.instanceStats.set(instanceId, event.stats);
}
user.recalculatePlayerStats();
this._controller.userUpdated(user);
let instance = this._controller.instances.get(instanceId);
let instance = this._controller.instances!.get(instanceId);
await lib.invokeHook(this._controller.plugins, "onPlayerEvent", instance, {
type: event.type,
name: event.name,
@@ -264,5 +291,3 @@ class HostConnection extends BaseConnection {
});
}
}
module.exports = HostConnection;
packages/controller/src/HttpCloser.js → packages/controller/src/HttpCloser.ts
+18 -16
View File
@@ -1,6 +1,7 @@
"use strict";
const http = require("http");
const https = require("https");
import type { Socket } from "net";
import http from "http";
import https from "https";
/**
* Gracefully close a Node.js HTTP(S) server
@@ -19,34 +20,37 @@ const https = require("https");
*
* @alias module:controller/src/HttpCloser
*/
class HttpCloser {
export default class HttpCloser {
private _server: http.Server | https.Server;
private _sockets: Set<Socket>;
private _responses: Set<http.ServerResponse>;
/**
* Server to attach this closer to.
* @param {https.Server | http.Server} server - HTTP(S) to attach to
*/
constructor(server) {
constructor(server: http.Server | https.Server) {
this._server = server;
this._sockets = new Set();
this._responses = new Set();
if (this._server instanceof http.Server) {
this._server.on("connection", this._addSocket.bind(this));
} else if (this._server instanceof https.Server) {
if (this._server instanceof https.Server) {
this._server.on("secureConnection", this._addSocket.bind(this));
} else if (this._server instanceof http.Server) {
this._server.on("connection", this._addSocket.bind(this));
} else {
throw Error("server must be an instance of either http.Server or https.Server.");
}
this._server.on("request", this._addRequest.bind(this));
}
_addSocket(socket) {
_addSocket(socket:Socket) {
this._sockets.add(socket);
socket.once("close", () => {
this._sockets.delete(socket);
});
}
_addRequest(request, response) {
_addRequest(request: http.IncomingMessage, response: http.ServerResponse) {
this._responses.add(response);
response.once("close", () => this._responses.delete(response));
response.once("finish", () => this._responses.delete(response));
@@ -62,16 +66,16 @@ class HttpCloser {
* If there's requests in progress that still haven't been completed
* after the timeout value passed they will be aborted.
*
* @param {number} timeout -
* @param timeout -
* Time in ms to wait for connections to close.
*/
async close(timeout = 5000) {
async close(timeout:number = 5000): Promise<void> {
let activeSockets = new Set();
for (let response of this._responses) {
if (!response.headersSent) {
response.setHeader("Connection", "close");
} else {
let socket = response.socket;
let socket = response.socket!;
response.once("finish", () => {
socket.end();
});
@@ -104,5 +108,3 @@ class HttpCloser {
});
}
};
module.exports = HttpCloser;
packages/controller/src/InstanceInfo.js
-47
View File
@@ -1,47 +0,0 @@
"use strict";
/**
* Runtime status of an instance on the controller
* @alias module:controller/src/InstanceInfo
*/
class InstanceInfo {
/** @type {module:lib.InstanceConfig} */
config;
/**
* Current status of the instance. One of:
* - `unknown`: Instance is assigned to a host but this host is currently
* not connected to the contreller.
* - `unassigned`: Instance is not assigned to a a host and exists only on
* the controller.
* - `stopped`: Instance is stopped.
* - `starting`: Instance is in the process of starting up.
* - `running`: Instance is running normally.
* - `stopping`: Instance is in the process of stopping.
* - `creating_save`: Instance is in the process of creating a save.
* - `exporting_data`: Instance is in the process of exporting game data.
* - `deleted`: Instance has been deleted.
*
* @type {string}
*/
status;
constructor({ config, status }) {
this.config = config;
this.status = status;
}
toJSON() {
return {
config: this.config.serialize(),
status: this.status,
};
}
/** Shorthand for `instance.config.get("instance.id")` @type {number} */
get id() {
return this.config.get("instance.id");
}
}
module.exports = InstanceInfo;
packages/controller/src/InstanceInfo.ts
+50
View File
@@ -0,0 +1,50 @@
import type * as lib from "@clusterio/lib";
/**
* Current status of the instance. One of:
* - `unknown`: Instance is assigned to a host but this host is currently
* not connected to the contreller.
* - `unassigned`: Instance is not assigned to a a host and exists only on
* the controller.
* - `stopped`: Instance is stopped.
* - `starting`: Instance is in the process of starting up.
* - `running`: Instance is running normally.
* - `stopping`: Instance is in the process of stopping.
* - `creating_save`: Instance is in the process of creating a save.
* - `exporting_data`: Instance is in the process of exporting game data.
* - `deleted`: Instance has been deleted.
*/
export type InstanceStatus =
"unknown" | "unassigned" | "stopped" | "starting" | "running" |
"stopping" | "creating_save" | "exporting_data" | "deleted";
/**
* Runtime status of an instance on the controller
* @alias module:controller/src/InstanceInfo
*/
export default class InstanceInfo {
config: lib.InstanceConfig;
status: InstanceStatus;
game_port: number | null = null;
constructor(
{ config, status }:
{ config: lib.InstanceConfig, status: InstanceStatus }
) {
this.config = config;
this.status = status;
}
toJSON(): Object {
return {
config: this.config.serialize(),
status: this.status,
};
}
/** Shorthand for `instance.config.get("instance.id")` */
get id():number {
return this.config.get("instance.id");
}
}
packages/controller/src/UserManager.js → packages/controller/src/UserManager.ts
+31 -37
View File
@@ -1,51 +1,47 @@
"use strict";
const fs = require("fs-extra");
const jwt = require("jsonwebtoken");
import fs from "fs-extra";
import jwt from "jsonwebtoken";
const lib = require("@clusterio/lib");
import * as lib from "@clusterio/lib";
/**
* Manages users and roles
* @alias module:controller/src/UserManager
*/
class UserManager {
constructor(config) {
this._config = config;
this.roles = null;
this.users = null;
export default class UserManager {
roles: Map<number, lib.Role> = new Map();
users: Map<string, lib.User> = new Map();
constructor(
private _config: lib.ControllerConfig
) {
}
async load(filePath) {
let loadedRoles = new Map();
let loadedUsers = new Map();
async load(filePath: string): Promise<void> {
try {
let content = JSON.parse(await fs.readFile(filePath));
let content = JSON.parse(await fs.readFile(filePath, { encoding: 'utf8' }));
for (let serializedRole of content.roles) {
let role = new lib.Role(serializedRole);
loadedRoles.set(role.id, role);
this.roles.set(role.id, role);
}
for (let serializedUser of content.users) {
let user = new lib.User(serializedUser, loadedRoles);
loadedUsers.set(user.name, user);
let user = new lib.User(serializedUser, this.roles);
this.users.set(user.name, user);
}
} catch (err) {
} catch (err: any) {
if (err.code !== "ENOENT") {
throw err;
}
// Create default roles if loading failed
lib.ensureDefaultAdminRole(loadedRoles);
lib.ensureDefaultPlayerRole(loadedRoles);
lib.ensureDefaultAdminRole(this.roles);
lib.ensureDefaultPlayerRole(this.roles);
}
this.roles = loadedRoles;
this.users = loadedUsers;
}
async save(filePath) {
if (!this.roles || !this.users) {
async save(filePath:string): Promise<void> {
if (this.roles.size === 0 || this.users.size === 0) {
return;
}
@@ -67,20 +63,21 @@ class UserManager {
}
/**
* 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.User} newly created user.
* @param name - Name of the user to create.
*/
createUser(name) {
createUser(name:string): lib.User {
if (this.users.has(name)) {
throw new Error(`User '${name}' already exists`);
}
let roles = [];
let defaultRoleId = this._config.get("controller.default_role_id");
let user = new lib.User({ name, roles: [defaultRoleId] }, this.roles);
if (defaultRoleId !== null) {
roles.push(defaultRoleId)
}
let user = new lib.User({ name, roles }, this.roles);
this.users.set(name, user);
return user;
}
@@ -88,16 +85,13 @@ class UserManager {
/**
* Sign access token for the given user name
*
* @param {string} name - user name to sign token for
* @returns {string} JWT access token for the user.
* @param name - user name to sign token for
* @returns JWT access token for the user.
*/
signUserToken(name) {
signUserToken(name: string): string {
return jwt.sign(
{ aud: "user", user: name },
Buffer.from(this._config.get("controller.auth_secret"), "base64")
);
}
}
module.exports = UserManager;
packages/controller/src/WsServer.js → packages/controller/src/WsServer.ts
+70 -55
View File
@@ -1,15 +1,19 @@
"use strict";
const jwt = require("jsonwebtoken");
const WebSocket = require("ws");
import type { IncomingMessage } from "http";
import type { Duplex } from "stream";
import type Controller from "./Controller";
import type { WebSocketClusterio } from "./WsServerConnector";
const lib = require("@clusterio/lib");
import jwt from "jsonwebtoken";
import WebSocket, { WebSocketServer } from "ws";
import * as lib from "@clusterio/lib";
const { logger } = lib;
const ControlConnection = require("./ControlConnection");
const packageVersion = require("../package").version;
const HostConnection = require("./HostConnection");
const WsServerConnector = require("./WsServerConnector");
import ControlConnection from "./ControlConnection";
import HostConnection, { HostInfo } from "./HostConnection";
import WsServerConnector from "./WsServerConnector";
const packageVersion = require("../package.json").version;
const wsMessageCounter = new lib.Counter(
"clusterio_controller_websocket_message_total",
@@ -31,35 +35,30 @@ const wsRejectedConnectionsCounter = new lib.Counter(
* WebSocket server
* @alias module:controller/src/WsServer
*/
class WsServer {
constructor(controller) {
export default class WsServer {
controller: Controller;
stopAcceptingNewSessions: boolean = false;
controlConnections: Map<number, ControlConnection> = new Map();
hostConnections: Map<number, HostConnection> = new Map();
activeConnectors: Map<number, WsServerConnector> = new Map();
pendingSockets: Set<WebSocket> = new Set();
// Unique string for the session token audience
sessionAud: string = `session-${Date.now()}`;
nextSessionId: number = 1;
nextControlId: number = 1;
wss: WebSocket.Server;
constructor(controller: Controller) {
this.controller = controller;
this.stopAcceptingNewSessions = false;
/** @type {Array<module:controller/src/ControlConnection>} */
this.controlConnections = new Map();
/** @type {Map<number, module:controller/src/HostConnection>} */
this.hostConnections = new Map();
/** @type {Map<number, module:controller/src/WsServerConnector>} */
this.activeConnectors = new Map();
/** @type {Set<module:ws>} */
this.pendingSockets = new Set();
// Unique string for the session token audience
this.sessionAud = `session-${Date.now()}`;
this.nextSessionId = 1;
this.nextControlId = 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));
this.wss.on("connection", (socket: WebSocketClusterio, req: IncomingMessage) => this.handleConnection(socket, req));
}
async stop() {
async stop(): Promise<void> {
this.stopAcceptingNewSessions = true;
let disconnectTasks = [];
@@ -75,7 +74,7 @@ class WsServer {
for (let task of disconnectTasks) {
try {
await task;
} catch (err) {
} catch (err: any) {
if (!(err instanceof lib.SessionLost)) {
logger.warn(`Unexpected error disconnecting connector:\n${err.stack}`);
}
@@ -87,7 +86,11 @@ class WsServer {
}
}
handleUpgrade(req, socket, head) {
handleUpgrade(
req: IncomingMessage,
socket: Duplex,
head: Buffer
) {
// For reasons that defy common sense, the connection event has
// to be emitted explictly when using noServer.
this.wss.handleUpgrade(req, socket, head, (ws) => {
@@ -95,7 +98,7 @@ class WsServer {
});
}
handleConnection(socket, req) {
handleConnection(socket: WebSocketClusterio, req: IncomingMessage) {
logger.verbose(`WsServer | new connection from ${req.socket.remoteAddress}`);
// Track statistics
@@ -112,11 +115,12 @@ class WsServer {
if (typeof args[0] === "string" && !socket.clusterio_ignore_dump) {
this.controller.debugEvents.emit("message", { direction: "out", content: args[0] });
}
// @ts-ignore TODO::: Issue with originalSend having multiple signatures ?
return originalSend.call(socket, ...args);
};
// Start connection handshake.
let loadedPlugins = {};
let loadedPlugins: any = {};
for (let [name, plugin] of this.controller.plugins) {
loadedPlugins[name] = plugin.info.version;
}
@@ -125,7 +129,10 @@ class WsServer {
this.attachHandler(socket, req);
}
attachHandler(socket, req) {
attachHandler(
socket: WebSocket,
req: IncomingMessage
) {
this.pendingSockets.add(socket);
let timeoutId = setTimeout(() => {
@@ -135,7 +142,7 @@ class WsServer {
this.pendingSockets.delete(socket);
}, 30*1000);
socket.once("message", (message) => {
socket.once("message", (message: string) => {
clearTimeout(timeoutId);
this.pendingSockets.delete(socket);
this.handleHandshake(
@@ -154,13 +161,13 @@ ${err.stack}`
});
}
async handleHandshake(rawMessage, socket, req) {
async handleHandshake(rawMessage: string, socket: WebSocket, req: IncomingMessage) {
let message;
try {
const json = JSON.parse(rawMessage);
lib.Message.validate(json);
message = lib.Message.fromJSON(json);
} catch (err) {
} catch (err: any) {
logger.verbose(`WsServer | closing ${req.socket.remoteAddress} after receiving invalid message`);
wsRejectedConnectionsCounter.inc();
socket.close(1002, `Invalid message: ${err.message}`);
@@ -168,7 +175,7 @@ ${err.stack}`
}
if (message.type === "resume") {
this.resumeSession(message.data, socket, req);
this.resumeSession((message as lib.MessageResume).data, socket, req);
return;
}
@@ -187,19 +194,19 @@ ${err.stack}`
}
if (message.type === "registerHost") {
await this.registerHost(message.data, socket, req);
await this.registerHost((message as lib.MessageRegisterHost).data, socket, req);
return;
}
if (message.type === "registerControl") {
await this.registerControl(message.data, socket, req);
await this.registerControl((message as lib.MessageRegisterControl).data, socket, req);
return;
}
throw new Error("This should be unreachable");
}
createSession(dst) {
createSession(dst: lib.Address): [ WsServerConnector, string ] {
let sessionId = this.nextSessionId;
this.nextSessionId += 1;
let sessionToken = jwt.sign(
@@ -217,7 +224,7 @@ ${err.stack}`
return [connector, sessionToken];
}
async registerHost(data, socket, req) {
async registerHost(data: lib.RegisterHostData, socket: WebSocket, req: IncomingMessage) {
try {
let tokenPayload = jwt.verify(
data.token,
@@ -226,11 +233,15 @@ ${err.stack}`
{ audience: ["host", "slave"] }
);
if (typeof tokenPayload === "string") {
throw new Error("unexpected JsonWebToken type");
}
// migrate: allow pre-rename tokens issued to hosts before alpha-14
if ((tokenPayload.host !== undefined ? tokenPayload.host : tokenPayload.slave) !== data.id) {
throw new Error("missmatched host id");
}
} catch (err) {
} catch (err: any) {
logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`);
wsRejectedConnectionsCounter.inc();
socket.close(4003, `Authentication failed: ${err.message}`);
@@ -258,18 +269,18 @@ ${err.stack}`
connector.on("close", () => {
if (this.hostConnections.get(data.id) === connection) {
this.hostConnections.delete(data.id);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
this.controller.hostUpdated(this.controller.hosts!.get(data.id) as HostInfo);
} else {
logger.warn("Unlisted HostConnection closed");
}
});
this.hostConnections.set(data.id, connection);
this.controller.hostUpdated(this.controller.hosts.get(data.id));
this.controller.hostUpdated(this.controller.hosts!.get(data.id) as HostInfo);
let src = new lib.Address(lib.Address.host, data.id);
connector.ready(socket, src, sessionToken);
connector.ready(socket, src, sessionToken, undefined);
}
async registerControl(data, socket, req) {
async registerControl(data: lib.RegisterControlData, socket: WebSocket, req: IncomingMessage) {
let user;
try {
let tokenPayload = jwt.verify(
@@ -278,16 +289,20 @@ ${err.stack}`
{ audience: "user" }
);
if (typeof tokenPayload === "string") {
throw new Error("unexpected JsonWebToken type");
}
user = this.controller.userManager.users.get(tokenPayload.user);
if (!user) {
throw new Error("invalid user");
}
if (tokenPayload.iat < user.tokenValidAfter) {
if ((tokenPayload.iat || 0) < user.tokenValidAfter) {
throw new Error("invalid token");
}
user.checkPermission("core.control.connect");
} catch (err) {
} catch (err: any) {
logger.verbose(`WsServer | authentication failed for ${req.socket.remoteAddress}`);
wsRejectedConnectionsCounter.inc();
socket.close(4003, `Authentication failed: ${err.message}`);
@@ -317,7 +332,7 @@ ${err.stack}`
connector.ready(socket, src, sessionToken, account);
}
resumeSession(data, socket, req) {
resumeSession(data: lib.ResumeData, socket: WebSocket, req: IncomingMessage) {
let connector;
try {
let payload = jwt.verify(
@@ -326,6 +341,10 @@ ${err.stack}`
{ audience: this.sessionAud }
);
if (typeof payload === "string") {
throw new Error("unexpected JsonWebToken type");
}
connector = this.activeConnectors.get(payload.sid);
if (!connector) {
throw new Error();
@@ -337,10 +356,6 @@ ${err.stack}`
return;
}
connector.continue(socket, data.lastSeq);
connector.continue(socket, data.lastSeq as number);
}
}
module.exports = WsServer;
packages/controller/src/WsServerConnector.js → packages/controller/src/WsServerConnector.ts
+57 -36
View File
@@ -1,9 +1,14 @@
"use strict";
const events = require("events");
import type WebSocket from "ws";
const lib = require("@clusterio/lib");
import events from "events";
import * as lib from "@clusterio/lib";
const { logger } = lib;
export interface WebSocketClusterio extends WebSocket {
clusterio_ignore_dump?: boolean;
}
/**
* Connector for controller connections
@@ -11,15 +16,17 @@ const { logger } = lib;
* @extends module:lib.WebSocketBaseConnector
* @alias module:controller/src/WsServerConnector
*/
class WsServerConnector extends lib.WebSocketBaseConnector {
constructor(dst, sessionId, sessionTimeout, heartbeatInterval) {
super(new lib.Address(lib.Address.controller, 0), dst);
export default class WsServerConnector extends lib.WebSocketBaseConnector {
_timeoutId: NodeJS.Timeout | null = null;
declare _socket: WebSocketClusterio | null;
this._sessionTimeout = sessionTimeout;
this._socket = null;
this._sessionId = sessionId;
this._heartbeatInterval = heartbeatInterval;
this._timeoutId = null;
constructor(
dst: lib.Address,
private _sessionId: number,
private _sessionTimeout: number,
public _heartbeatInterval: number,
) {
super(new lib.Address(lib.Address.controller, 0), dst);
// The following states are used in the server connector
// closed: Connection is closed
@@ -28,8 +35,10 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
}
_reset() {
clearTimeout(this._timeoutId);
this._timeoutId = null;
if (this._timeoutId) {
clearTimeout(this._timeoutId);
this._timeoutId = null;
}
super._reset();
}
@@ -38,14 +47,17 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
*
* Sends the ready message over the socket to initiate the session.
*
* @param {Object} socket - WebSocket connection to client.
* @param {module:lib.Address} src - Source address for this link.
* @param {string} sessionToken -
* the session token to send to the client.
* @param {module:lib.AccountDetails=} account -
* account data to provide to control connection
* @param socket - WebSocket connection to client.
* @param src - Source address for this link.
* @param sessionToken - the session token to send to the client.
* @param account - account data to provide to control connection
*/
ready(socket, src, sessionToken, account) {
ready(
socket: WebSocket.WebSocket,
src: lib.Address,
sessionToken: string,
account: lib.AccountDetails | undefined
) {
this._socket = socket;
this._sendInternal(new lib.MessageReady(
new lib.ReadyData(
@@ -68,14 +80,17 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
* 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.
* @param socket - New socket to continue on.
* @param lastSeq - The last message the client received.
*/
continue(socket, lastSeq) {
continue(
socket: WebSocket.WebSocket,
lastSeq: number
) {
// It's possible the previous connection hasn't closed yet due to a
// stale connection. Terminate it if so.
if (this._state === "connected") {
if (this._state === "connected" && this._socket !== null) {
this._socket.terminate();
this._socket.once("close", () => this.continue(socket, lastSeq));
return;
@@ -108,9 +123,13 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
}
_attachSocketHandlers() {
if (this._socket === null) {
throw new Error("this._socket is null");
}
this.startHeartbeat();
this._socket.on("close", (code, reason) => {
this._socket.on("close", (code: number, reason: string) => {
logger.verbose(`Connector | Close (code: ${code}, reason: ${reason})`);
this.stopHeartbeat();
this._socket = null;
@@ -124,10 +143,9 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
this.emit("drop");
this._timeoutId = setTimeout(() => { this._timedOut(); }, this._sessionTimeout * 1000);
}
});
this._socket.on("error", err => {
this._socket.on("error", (err: any) => {
// It's assumed that close is always called by ws
logger.verbose("Connector | Error:", err);
});
@@ -135,15 +153,15 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
this._socket.on("open", () => {
logger.verbose("Connector | Open");
});
this._socket.on("ping", data => {
this._socket.on("ping", (data: string) => {
logger.verbose(`Connector | Ping (data: ${data}`);
});
this._socket.on("pong", data => {
this._socket.on("pong", (data: string) => {
logger.verbose(`Connector | Pong (data: ${data}`);
});
// Handle messages
this._socket.on("message", data => {
this._socket.on("message", (data: string) => {
let message = this._parseMessage(data);
if (!message) {
return;
@@ -162,10 +180,13 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
*
* Sends a close frame and disconnects the connector.
*
* @param {number} code - WebSocket close code.
* @param {string} reason - WebSocket close reason.
* @param code - WebSocket close code.
* @param reason - WebSocket close reason.
*/
async close(code, reason) {
async close(
code: number,
reason: string
): Promise<void> {
if (this._state === "closed") {
return;
}
@@ -177,7 +198,9 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
}
this._closing = true;
this._socket.close(code, reason);
if (this._socket !== null) {
this._socket.close(code, reason);
}
await events.once(this, "close");
}
@@ -213,5 +236,3 @@ class WsServerConnector extends lib.WebSocketBaseConnector {
this._closing = true;
}
}
module.exports = WsServerConnector;
packages/controller/src/metrics.js
-14
View File
@@ -1,14 +0,0 @@
// Shared metrics
"use strict";
const lib = require("@clusterio/lib");
const util = require("util");
exports.endpointHitCounter = new lib.Counter(
"clusterio_controller_http_endpoint_hits_total",
"How many requests a particular HTTP endpoint has gotten",
{ labels: ["route"], register: false }
);
exports.endpointHitCounter.labels = util.deprecate(
exports.endpointHitCounter.labels, "incrementing endpointHitCounter is no longer needed and has no effect"
);
packages/controller/src/metrics.ts
+15
View File
@@ -0,0 +1,15 @@
import * as lib from "@clusterio/lib";
import util from "util";
const endpointHitCounter = new lib.Counter(
"clusterio_controller_http_endpoint_hits_total",
"How many requests a particular HTTP endpoint has gotten",
{ labels: ["route"], register: false }
);
endpointHitCounter.labels = util.deprecate(
endpointHitCounter.labels,
"incrementing endpointHitCounter is no longer needed and has no effect"
);
export default endpointHitCounter
packages/controller/src/routes.js → packages/controller/src/routes.ts
+111 -86
View File
@@ -1,26 +1,28 @@
"use strict";
const busboy = require("busboy");
const crypto = require("crypto");
const events = require("events");
const fs = require("fs-extra");
const JSZip = require("jszip");
const jwt = require("jsonwebtoken");
const path = require("path");
const nodeStream = require("stream");
const util = require("util");
import type { Application, Request, Response } from "express";
import type Controller from "./Controller";
const lib = require("@clusterio/lib");
import busboy from "busboy";
import crypto from "crypto";
import events from "events";
import fs from "fs-extra";
import JSZip from "jszip";
import jwt from "jsonwebtoken";
import path from "path";
import nodeStream from "stream";
import util from "util";
import * as lib from "@clusterio/lib";
const { logger } = lib;
const finished = util.promisify(nodeStream.finished);
// Merges samples from sourceResult to destinationResult
function mergeSamples(destinationResult, sourceResult) {
function mergeSamples(destinationResult: any, sourceResult: any) {
let receivedSamples = new Map(sourceResult.samples);
for (let [suffix, suffixSamples] of destinationResult.samples) {
if (receivedSamples.has(suffix)) {
suffixSamples.push(...receivedSamples.get(suffix));
suffixSamples.push(...receivedSamples.get(suffix) as any[]);
receivedSamples.delete(suffix);
}
}
@@ -31,23 +33,25 @@ function mergeSamples(destinationResult, sourceResult) {
}
// Prometheus polling endpoint
async function getMetrics(req, res, next) {
let results = [];
let pluginResults = await lib.invokeHook(req.app.locals.controller.plugins, "onMetrics");
async function getMetrics(req: Request, res: Response, next: any) {
const controller: Controller = req.app.locals.controller
let results: any[] = [];
let pluginResults: any[] = await lib.invokeHook(controller.plugins, "onMetrics");
for (let metricIterator of pluginResults) {
for await (let metric of metricIterator) {
results.push(metric);
}
}
let requests = [];
let timeout = req.app.locals.controller.config.get("controller.metrics_timeout") * 1000;
for (let [hostId, hostConnection] of req.app.locals.controller.wsServer.hostConnections) {
let requests: Promise<any>[] = [];
let timeout = controller.config.get("controller.metrics_timeout") * 1000;
for (let [hostId, hostConnection] of controller.wsServer.hostConnections) {
if (!hostConnection.connected) {
continue;
}
requests.push(lib.timeout(
hostConnection.send(new lib.HostMetricsRequest()).catch(err => {
hostConnection.send(new lib.HostMetricsRequest()).catch((err: any) => {
if (!(err instanceof lib.SessionLost)) {
logger.error(`Unexpected error gathering metrics from host:\n${err.stack}`);
}
@@ -61,7 +65,7 @@ async function getMetrics(req, res, next) {
results.push(result);
}
let resultMap = new Map();
let resultMap: Map<string, any> = new Map();
for (let response of await Promise.all(requests)) {
if (!response) {
// TODO: Log timeout occured?
@@ -74,28 +78,34 @@ async function getMetrics(req, res, next) {
} else {
// Merge metrics received by multiple hosts
mergeSamples(resultMap.get(result.metric.name), result);
mergeSamples(resultMap.get(result.metric.name)!, result);
}
}
}
// TODO: results seem to mix deserialized and serialized lib.CollectorResult ?
// mergeSamples imply samples be [string, [string, number][]]
// but lib.defaultRegistry.collect() imply Map<string, Map<string, number>>
for (let result of resultMap.values()) {
results.push(lib.deserializeResult(result));
}
// lib.exposition expect AsyncIterable<CollectorResult> as parameter.
// Should wrap results into a Generator functions ?
// @ts-ignore
let text = await lib.exposition(results);
res.set("Content-Type", lib.exposition.contentType);
res.send(text);
}
function getPlugins(req, res) {
function getPlugins(req: Request, res: Response) {
let plugins = [];
for (let pluginInfo of req.app.locals.controller.pluginInfos) {
let name = pluginInfo.name;
let loaded = req.app.locals.controller.plugins.has(name);
let enabled = loaded && req.app.locals.controller.config.group(pluginInfo.name).get("load_plugin");
let web = {};
let web: { main:any, error:string|undefined } = { main: undefined, error: undefined};
let devPlugins = req.app.locals.devPlugins;
if (devPlugins && devPlugins.has(name)) {
let stats = res.locals.webpack.devMiddleware.stats.stats[devPlugins.get(name)];
@@ -116,7 +126,7 @@ function getPlugins(req, res) {
res.send(plugins);
}
function validateHostToken(req, res, next) {
function validateHostToken(req: Request, res: Response, next: any) {
let token = req.header("x-access-token");
if (!token) {
res.sendStatus(401);
@@ -138,7 +148,7 @@ function validateHostToken(req, res, next) {
next();
}
function validateUserToken(req, res, next) {
function validateUserToken(req: Request, res: Response, next: any) {
let token = req.header("x-access-token");
if (!token) {
res.sendStatus(401);
@@ -151,11 +161,15 @@ function validateUserToken(req, res, next) {
Buffer.from(req.app.locals.controller.config.get("controller.auth_secret"), "base64"),
{ audience: "user" }
);
if (typeof tokenPayload === "string") {
throw new Error("unexpected JsonWebToken type");
}
let user = req.app.locals.controller.userManager.users.get(tokenPayload.user);
if (!user) {
throw new Error("invalid user");
}
if (tokenPayload.iat < user.tokenValidAfter) {
if (!tokenPayload.iat || tokenPayload.iat < user.tokenValidAfter) {
throw new Error("invalid token");
}
res.locals.user = user;
@@ -169,7 +183,7 @@ function validateUserToken(req, res, next) {
}
// Handle an uploaded export package.
async function uploadExport(req, res) {
async function uploadExport(req: Request, res: Response) {
if (req.get("Content-Type") !== "application/zip") {
res.sendStatus(415);
return;
@@ -194,9 +208,9 @@ async function uploadExport(req, res) {
for await (let chunk of req) {
data.push(chunk);
}
data = Buffer.concat(data);
let zip = await JSZip.loadAsync(data);
data = null;
let buffer: Buffer | null = Buffer.concat(data);
let zip = await JSZip.loadAsync(buffer);
buffer = null;
// This is hardcoded to prevent path expansion attacks
let exportFiles = [
@@ -207,7 +221,7 @@ async function uploadExport(req, res) {
"export/locale.json",
];
let assets = {};
let assets: any = {};
let settingPrototypes = {};
for (let filePath of exportFiles) {
let file = zip.file(filePath);
@@ -232,11 +246,24 @@ async function uploadExport(req, res) {
res.sendStatus(200);
}
async function createProxyStream(app) {
interface ProxyStream {
id: string;
source: nodeStream.Readable | null;
flowing: boolean;
size: number | null;
mime: string | null;
filename: string | null;
events: events.EventEmitter;
timeout: NodeJS.Timeout;
}
export async function createProxyStream(app: Application): Promise<ProxyStream> {
let asyncRandomBytes = util.promisify(crypto.randomBytes);
let id = (await asyncRandomBytes(8)).toString("hex");
let stream = {
let stream: ProxyStream = {
id,
source: null,
flowing: false,
size: null,
mime: null,
@@ -257,7 +284,7 @@ async function createProxyStream(app) {
return stream;
}
async function putStream(req, res) {
async function putStream(req: Request, res: Response) {
let stream = req.app.locals.streams.get(req.params.id);
if (!stream || stream.source) {
res.sendStatus(404);
@@ -278,7 +305,7 @@ async function putStream(req, res) {
});
}
async function getStream(req, res) {
async function getStream(req: Request, res: Response) {
let stream = req.app.locals.streams.get(req.params.id);
if (!stream || stream.flowing) {
res.sendStatus(404);
@@ -321,15 +348,15 @@ const zipMimes = [
const contentTypeRegExp = /^([!#$%&'*+\-.^_`|~0-9A-Za-z]+\/[!#$%&'*+\-.^_`|~0-9A-Za-z]+)/;
async function uploadSave(req, res) {
async function uploadSave(req: Request, res: Response) {
try {
res.locals.user.checkPermission("core.instance.save.upload");
} catch (err) {
} catch (err: any) {
res.status(403).json({ request_errors: [err.message] });
return;
}
let contentType = req.get("Content-Type");
let contentType = req.get("Content-Type") || "";
let match = contentTypeRegExp.exec(contentType);
if (!match) {
res.status(415).json({ request_errors: ["invalid Content-Type"] });
@@ -338,11 +365,11 @@ async function uploadSave(req, res) {
let contentMime = match[1].toLowerCase();
let tasks = [];
let errors = [];
let errors: string[] = [];
let requestErrors = [];
let saves = [];
let saves: string[] = [];
async function handleFile(instanceId, stream, filename, streamMime) {
async function handleFile(instanceId:number, stream: nodeStream.Readable, filename: string, streamMime: string) {
let proxyStream = await createProxyStream(req.app);
proxyStream.source = stream;
proxyStream.mime = streamMime;
@@ -354,7 +381,7 @@ async function uploadSave(req, res) {
});
try {
let storedName = await Promise.race([
let storedName: string = await Promise.race([
req.app.locals.controller.sendToHostByInstanceId(
new lib.InstancePullSaveRequest(
instanceId,
@@ -366,7 +393,7 @@ async function uploadSave(req, res) {
]);
saves.push(storedName);
} catch (err) {
} catch (err: any) {
proxyStream.events.emit("close");
logger.error(`Error uploading save: ${err.message}`);
errors.push(err.message);
@@ -376,10 +403,10 @@ async function uploadSave(req, res) {
if (contentMime === "multipart/form-data") {
await new Promise(resolve => {
let fields = {};
let instanceId:number | undefined;
let parser = busboy({ headers: req.headers });
parser.on("file", (name, stream, { filename, mimeType }) => {
if (fields.instanceId === undefined) {
parser.on("file", (name: string, stream: nodeStream.Readable, { filename, mimeType }: { filename: string, mimeType: string } ) => {
if (instanceId === undefined) {
requestErrors.push("instance_id must come before files uploaded");
}
@@ -396,18 +423,18 @@ async function uploadSave(req, res) {
return;
}
tasks.push(handleFile(fields.instanceId, stream, filename, mimeType));
tasks.push(handleFile(instanceId as number, stream, filename, mimeType));
});
parser.on("field", (name, value, info) => {
if (name === "instance_id") {
fields.instanceId = Number.parseInt(value, 10);
if (Number.isNaN(fields.instanceId)) {
instanceId = Number.parseInt(value, 10);
if (Number.isNaN(instanceId)) {
requestErrors.push("invalid instance_id");
}
}
});
parser.on("close", resolve);
parser.on("error", (err) => {
parser.on("error", (err: any) => {
logger.error(`Error parsing multipart request in upload-save:\n${err.stack}`);
errors.push(err.message);
});
@@ -421,7 +448,7 @@ async function uploadSave(req, res) {
} else if (!filename.endsWith(".zip")) {
requestErrors.push("filename must end with .zip");
}
let instanceId = Number.parseInt(req.query.instance_id, 10);
let instanceId = Number.parseInt(String(req.query.instance_id), 10);
if (Number.isNaN(instanceId)) {
requestErrors.push("Missing or invalid instance_id parameter");
}
@@ -429,7 +456,7 @@ async function uploadSave(req, res) {
if (errors.length || requestErrors.length) {
req.resume();
} else {
tasks.push(handleFile(instanceId, req, filename, contentMime));
tasks.push(handleFile(instanceId, req, filename as string, contentMime));
}
} else {
@@ -453,23 +480,23 @@ async function uploadSave(req, res) {
res.json({ saves });
}
function checkModName(name) {
function checkModName(name: string) {
try {
lib.checkFilename(name);
} catch (err) {
} catch (err: any) {
throw new lib.RequestError(`Mod name ${err.message}`);
}
}
async function uploadMod(req, res) {
async function uploadMod(req: Request, res: Response) {
try {
res.locals.user.checkPermission("core.mod.upload");
} catch (err) {
} catch (err: any) {
res.status(403).json({ request_errors: [err.message] });
return;
}
let contentType = req.get("Content-Type");
let contentType = req.get("Content-Type")!;
let match = contentTypeRegExp.exec(contentType);
if (!match) {
res.status(415).json({ request_errors: ["invalid Content-Type"] });
@@ -478,15 +505,15 @@ async function uploadMod(req, res) {
let contentMime = match[1].toLowerCase();
let tasks = [];
let errors = [];
let errors: string[] = [];
let requestErrors = [];
let mods = [];
let mods: any[] = [];
async function handleFile(stream, filename) {
async function handleFile(stream: nodeStream.Readable, filename: string) {
try {
checkModName(filename);
} catch (err) {
} catch (err: any) {
logger.error(`Error uploading mod: ${err.message}`);
errors.push(err.message);
stream.resume();
@@ -495,15 +522,16 @@ async function uploadMod(req, res) {
const modsDirectory = req.app.locals.controller.config.get("controller.mods_directory");
let tempFilename = filename.replace(/(\.zip)?$/, ".tmp.zip");
let writeStream;
try {
let writeStream;
while (true) {
try {
writeStream = fs.createWriteStream(path.join(modsDirectory, tempFilename), { flags: "wx" });
await events.once(writeStream, "open");
break;
} catch (err) {
} catch (err: any) {
if (err.code === "EEXIST") {
tempFilename = await lib.findUnusedName(modsDirectory, tempFilename, ".tmp.zip");
} else {
@@ -520,18 +548,21 @@ async function uploadMod(req, res) {
req.app.locals.controller.modUpdated(modInfo);
mods.push(modInfo.toJSON());
} catch (err) {
} catch (err: any) {
logger.error(`Error uploading mod: ${err.message}`);
errors.push(err.message);
stream.resume();
// Attempt to clean up.
writeStream.destroy();
if (writeStream) {
writeStream.destroy();
}
try {
await fs.unlink(path.join(modsDirectory, tempFilename));
} catch (unlinkErr) {
} catch (unlinkErr: any) {
if (unlinkErr.code !== "ENOENT") {
logger.error(`Error removing ${tempFilename}: ${err.message}`);
logger.error(`Error removing ${tempFilename}: ${unlinkErr.message}`);
}
}
}
@@ -540,7 +571,7 @@ async function uploadMod(req, res) {
if (contentMime === "multipart/form-data") {
await new Promise(resolve => {
let parser = busboy({ headers: req.headers });
parser.on("file", (name, stream, { filename, mimeType }) => {
parser.on("file", (name:string, stream: nodeStream.Readable, { filename, mimeType }: { filename: string, mimeType: string }) => {
if (!zipMimes.includes(mimeType)) {
requestErrors.push("invalid file Content-Type");
}
@@ -557,7 +588,7 @@ async function uploadMod(req, res) {
tasks.push(handleFile(stream, filename));
});
parser.on("close", resolve);
parser.on("error", (err) => {
parser.on("error", (err: any) => {
logger.error(`Error parsing multipart request in upload-mod:\n${err.stack}`);
errors.push(err.message);
});
@@ -575,7 +606,7 @@ async function uploadMod(req, res) {
if (errors.length || requestErrors.length) {
req.resume();
} else {
tasks.push(handleFile(req, filename));
tasks.push(handleFile(req, filename as string));
}
} else {
@@ -600,27 +631,27 @@ async function uploadMod(req, res) {
}
function addRouteHandlers(app) {
app.get("/metrics", (req, res, next) => getMetrics(req, res, next).catch(next));
export function addRouteHandlers(app: Application) {
app.get("/metrics", (req:Request, res:Response, next:any) => getMetrics(req, res, next).catch(next));
app.get("/api/plugins", getPlugins);
app.put("/api/upload-export",
validateHostToken,
(req, res, next) => uploadExport(req, res).catch(next)
(req:Request, res:Response, next:any) => uploadExport(req, res).catch(next)
);
app.put("/api/stream/:id", (req, res, next) => putStream(req, res).catch(next));
app.get("/api/stream/:id", (req, res, next) => getStream(req, res).catch(next));
app.put("/api/stream/:id", (req:Request, res:Response, next:any) => putStream(req, res).catch(next));
app.get("/api/stream/:id", (req:Request, res:Response, next:any) => getStream(req, res).catch(next));
app.post("/api/upload-save",
validateUserToken,
(req, res, next) => uploadSave(req, res).catch(next)
(req:Request, res:Response, next:any) => uploadSave(req, res).catch(next)
);
app.post("/api/upload-mod",
validateUserToken,
(req, res, next) => uploadMod(req, res).catch(next)
(req:Request, res:Response, next:any) => uploadMod(req, res).catch(next)
);
}
// Routes used in the web interface and served by the controller
const webRoutes = [
export const webRoutes = [
"/",
"/controller",
"/hosts",
@@ -636,9 +667,3 @@ const webRoutes = [
"/plugins",
"/plugins/:name/view",
];
module.exports = {
addRouteHandlers,
webRoutes,
createProxyStream,
};
packages/controller/tsconfig.json
+17
View File
@@ -0,0 +1,17 @@
{
"include": ["controller.ts", "package.json", "src/**/*.ts"],
"compilerOptions": {
"lib": ["es2023"],
"module": "commonjs",
"target": "es2022",
"rootDir": ".",
"outDir": "dist/controller",
"composite": true,
"sourceMap": true,
"strict": true,
"alwaysStrict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"resolveJsonModule": true
}
}
packages/controller/typedoc.json
+4
View File
@@ -0,0 +1,4 @@
{
"entryPoints": ["controller.ts"],
"includeVersion": true
}
packages/lib/src/config/classes.ts
+1 -1
View File
@@ -58,7 +58,7 @@ function splitOn(separator: string, string: string) {
return [string.slice(0, index), string.slice(index + separator.length)];
}
type SerializedConfig = ReturnType<Config["serialize"]>;
export type SerializedConfig = ReturnType<Config["serialize"]>;
type SerializedGroup = ReturnType<ConfigGroup["serialize"]>;
/**
packages/lib/src/config/definitions.ts
+62 -3
View File
@@ -140,7 +140,27 @@ ControllerGroup.finalize();
* Controller Config class
* @extends classes.Config
*/
export class ControllerConfig extends classes.Config { }
export class ControllerConfig extends classes.Config {
get(name: "controller.name"): string;
get(name: "controller.mods_directory"): string;
get(name: "controller.database_directory"): string;
get(name: "controller.http_port"): number | null;
get(name: "controller.https_port"): number | null;
get(name: "controller.bind_address"): string | null;
get(name: "controller.external_address"): string | null;
get(name: "controller.tls_certificate"): string | null;
get(name: "controller.tls_private_key"): string | null;
get(name: "controller.auth_secret"): string;
get(name: "controller.heartbeat_interval"): number;
get(name: "controller.session_timeout"): number;
get(name: "controller.metrics_timeout"): number;
get(name: "controller.proxy_stream_timeout"): number;
get(name: "controller.default_mod_pack_id"): number | null;
get(name: "controller.default_role_id"): number | null;
get(name: string) {
return super.get(name);
}
}
ControllerConfig.registerGroup(ControllerGroup);
@@ -230,7 +250,22 @@ HostGroup.finalize();
* Host Config class
* @extends classes.Config
*/
export class HostConfig extends classes.Config { }
export class HostConfig extends classes.Config {
get(name: "host.name"): string;
get(name: "host.id"): number;
get(name: "host.factorio_directory"): string;
get(name: "host.mods_directory"): string;
get(name: "host.mods_directory_is_shared"): boolean;
get(name: "host.instances_directory"): string;
get(name: "host.controller_url"): string;
get(name: "host.controller_token"): string;
get(name: "host.tls_ca"): string | null;
get(name: "host.public_address"): string;
get(name: "host.max_reconnect_delay"): number;
get(name: string) {
return super.get(name);
}
}
HostConfig.registerGroup(HostGroup);
@@ -393,7 +428,31 @@ FactorioGroup.finalize();
* Instance config class
* @extends classes.Config
*/
export class InstanceConfig extends classes.Config { }
export class InstanceConfig extends classes.Config {
get(name: "instance.name"): string;
get(name: "instance.id"): number;
get(name: "instance.assigned_host"): number | null;
get(name: "instance.auto_start"): boolean;
get(name: "factorio.version"): string;
get(name: "factorio.game_port"): number | null;
get(name: "factorio.rcon_port"): number | null;
get(name: "factorio.rcon_password"): string | null;
get(name: "factorio.player_online_autosave_slots"): number;
get(name: "factorio.mod_pack"): string | null;
get(name: "factorio.enable_save_patching"): boolean;
get(name: "factorio.enable_whitelist"): boolean;
get(name: "factorio.enable_authserver_bans"): boolean;
get(name: "factorio.settings"): Object;
get(name: "factorio.verbose_logging"): boolean;
get(name: "factorio.strip_paths"): boolean;
get(name: "factorio.sync_adminlist"): boolean;
get(name: "factorio.sync_whitelist"): boolean;
get(name: "factorio.max_concurrent_commands"): number;
get(name: string) {
return super.get(name);
}
}
InstanceConfig.registerGroup(InstanceGroup);
InstanceConfig.registerGroup(FactorioGroup);
packages/lib/src/data/messages_controller.ts
+1 -1
View File
@@ -202,7 +202,7 @@ export class LogMessageEvent {
static dst = ["controller", "control"] as const;
constructor(
public info: object,
public info: { level: string, message: string },
) { }
static jsonSchema = Type.Object({
packages/lib/src/data/messages_core.ts
+2
View File
@@ -172,6 +172,8 @@ export class Message {
}
}
export type MessageSrcDst = MessageRequest | MessageResponse | MessageResponseError | MessageEvent;
export class HelloData {
constructor(
public version: string,
packages/lib/src/link/link.ts
+4 -4
View File
@@ -37,7 +37,7 @@ export interface Event<T> {
}
export type EventClass<T> = Event<T>["constructor"];
interface RequestEntry {
export interface RequestEntry {
Request: RequestClass<unknown, unknown>;
name: string,
allowedSrcTypes: Set<number>;
@@ -60,15 +60,15 @@ interface ForwardedRequest {
dst: libData.Address;
}
interface EventEntry {
export interface EventEntry {
Event: EventClass<unknown>;
name: string,
allowedSrcTypes: Set<number>;
allowedDstTypes: Set<number>;
eventFromJSON: (json: any) => Event<unknown>;
}
type RequestHandler<Req, Res> = (request: Req, src: libData.Address, dst: libData.Address) => Promise<Res>;
type EventHandler<T> = (event: T, src: libData.Address, dst: libData.Address) => Promise<void>;
export type RequestHandler<Req, Res> = (request: Req, src: libData.Address, dst: libData.Address) => Promise<Res>;
export type EventHandler<T> = (event: T, src: libData.Address, dst: libData.Address) => Promise<void>;
interface Router {
forwardMessage(origin: Link, message: Message, fallback: boolean): boolean,
packages/lib/src/logging_utils.ts
+2 -2
View File
@@ -421,7 +421,7 @@ export interface QueryLogFilter {
/**
* Maximum log level to include. Higher levels are more verbose.
*/
maxLevel: keyof typeof levels;
maxLevel?: keyof typeof levels;
/**
* Include log entries from controller, all hosts and all instances.
*/
@@ -451,7 +451,7 @@ export interface QueryLogFilter {
* Index to speed up query with.
* @returns log entries matching the filter
*/
export async function queryLog(logDirectory: string, filter: QueryLogFilter, index: LogIndex) {
export async function queryLog(logDirectory: string, filter: QueryLogFilter, index: LogIndex | null) {
let files = (await fs.readdir(logDirectory)).filter(entry => logFileGlob.test(entry));
filter = {
test/controller/HttpCloser.js
+1 -1
View File
@@ -7,7 +7,7 @@ const util = require("util");
const phin = require("phin");
const HttpCloser = require("@clusterio/controller/src/HttpCloser");
const HttpCloser = require("@clusterio/controller/dist/controller/src/HttpCloser").default;
// Time to wait during async operations to ensure they happened in order.
const tick = 20;
test/controller/routes.js
+1 -1
View File
@@ -7,7 +7,7 @@ const http = require("http");
const phin = require("phin");
const { wait } = require("@clusterio/lib");
const routes = require("@clusterio/controller/src/routes");
const routes = require("@clusterio/controller/dist/controller/src/routes");
const mock = require("../mock");
test/mock.js
+1 -1
View File
@@ -5,7 +5,7 @@ const express = require("express");
const lib = require("@clusterio/lib");
const UserManager = require("@clusterio/controller/src/UserManager");
const UserManager = require("@clusterio/controller/dist/controller/src/UserManager").default;
const addr = lib.Address.fromShorthand;