Merge pull request #533 from Hornwitser/sync-mods-with-controller

Sync mods with controller

.github/workflows/node.js.yml

@@ -26,8 +26,6 @@ jobs:
         node-version: ${{ matrix.node-version }}
     - run: pnpm i --no-optional --no-frozen-lockfile
     - run: wget -q -O factorio.tar.gz https://www.factorio.com/get-download/latest/headless/linux64 && tar -xf factorio.tar.gz && rm factorio.tar.gz
-    - run: node packages/lib/build_mod --source-dir packages/host/lua/clusterio_lib --output-dir temp/test/mods
-    - run: wget -q -O temp/test/mods/subspace_storage_1.99.8.zip -L https://github.com/clusterio/subspace_storage/releases/download/1.99.6/subspace_storage_1.99.8.zip
     - name: Run tests
       if: ${{ matrix.node-version != '18.x' }}
       run: pnpm test

packages/controller/src/BaseConnection.ts

@@ -1,9 +1,13 @@
+import path from "path";
+import fs from "fs-extra";
+
 import type Controller from "./Controller";
 import type WsServerConnector from "./WsServerConnector";
 
 import * as lib from "@clusterio/lib";
 const { logger } = lib;
 import ControllerRouter from "./ControllerRouter";
+import * as routes from "./routes";
 
 
 /**
@@ -13,7 +17,7 @@ import ControllerRouter from "./ControllerRouter";
  * @alias module:controller/src/BaseConnection
  */
 export default class BaseConnection extends lib.Link {
-	declare connector: lib.WebSocketBaseConnector;
+	declare ["connector"]: WsServerConnector;
 
 	constructor(
 		connector: lib.WebSocketBaseConnector,
@@ -28,6 +32,7 @@ export default class BaseConnection extends lib.Link {
 
 		this.handle(lib.ModPackGetRequest, this.handleModPackGetRequest.bind(this));
 		this.handle(lib.ModPackGetDefaultRequest, this.handleModPackGetDefaultRequest.bind(this));
+		this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this));
 	}
 
 	async disconnect(code: number, reason: string) {
@@ -68,4 +73,29 @@ export default class BaseConnection extends lib.Link {
 		}
 		return modPack;
 	}
+
+	getMod(mod: { name: string, version: string, sha1?: string }) {
+		let filename = lib.ModInfo.filename(mod.name, mod.version);
+		let modInfo = this._controller.modStore.files.get(filename);
+		if (!modInfo) {
+			throw new lib.RequestError(`Mod ${filename} does not exist on controller`);
+		}
+		if (mod.sha1 && mod.sha1 !== modInfo.sha1) {
+			throw new lib.RequestError(`Mod ${filename} checksum does not match controller's checksum`);
+		}
+		return modInfo;
+	}
+
+	async handleModDownloadRequest(request: lib.ModDownloadRequest) {
+		let mod = this.getMod(request);
+		let modPath = path.join(this._controller.config.get("controller.mods_directory"), mod.filename);
+
+		let stream = await routes.createProxyStream(this._controller.app);
+		stream.filename = mod.filename;
+		stream.source = fs.createReadStream(modPath);
+		stream.mime = "application/zip";
+		stream.size = String(mod.size);
+
+		return stream.id;
+	}
 }

packages/controller/src/ControlConnection.ts

@@ -2,8 +2,6 @@ import type WsServerConnector from "./WsServerConnector";
 import type { HostInfo } from "./HostConnection";
 
 import events from "events";
-import fs from "fs-extra";
-import path from "path";
 
 import * as lib from "@clusterio/lib";
 const { logFilter, logger } = lib;
@@ -88,9 +86,9 @@ export default class ControlConnection extends BaseConnection {
 		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.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.InstanceRenameSaveRequest, controller.sendRequestToHostByInstanceId.bind(controller));
+		this.handle(lib.InstanceCopySaveRequest, controller.sendRequestToHostByInstanceId.bind(controller));
+		this.handle(lib.InstanceDeleteSaveRequest, controller.sendRequestToHostByInstanceId.bind(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));
@@ -100,7 +98,6 @@ export default class ControlConnection extends BaseConnection {
 		this.handle(lib.ModGetRequest, this.handleModGetRequest.bind(this));
 		this.handle(lib.ModListRequest, this.handleModListRequest.bind(this));
 		this.handle(lib.ModSearchRequest, this.handleModSearchRequest.bind(this));
-		this.handle(lib.ModDownloadRequest, this.handleModDownloadRequest.bind(this));
 		this.handle(lib.ModDeleteRequest, this.handleModDeleteRequest.bind(this));
 		this.handle(lib.LogSetSubscriptionsRequest, this.handleLogSetSubscriptionsRequest.bind(this));
 		this.handle(lib.LogQueryRequest, this.handleLogQueryRequest.bind(this));
@@ -461,17 +458,11 @@ export default class ControlConnection extends BaseConnection {
 	}
 
 	async handleModGetRequest(request: lib.ModGetRequest) {
-		let { name, version } = request;
-		let filename = `${name}_${version}.zip`;
-		let mod = this._controller.mods.get(filename);
-		if (!mod) {
-			throw new lib.RequestError(`Mod ${filename} does not exist`);
-		}
-		return mod;
+		return this.getMod(request);
 	}
 
 	async handleModListRequest() {
-		return [...this._controller.mods.values()];
+		return [...this._controller.modStore.mods()];
 	}
 
 	static termsMatchesMod(terms: lib.ParsedTerm[], mod: lib.ModInfo) {
@@ -511,7 +502,7 @@ export default class ControlConnection extends BaseConnection {
 
 		type ModVersions = { name: string, versions: lib.ModInfo[] };
 		let results: Map<string, ModVersions> = new Map();
-		for (let mod of this._controller.mods.values()) {
+		for (let mod of this._controller.modStore.mods()) {
 			if (
 				mod.factorioVersion !== factorioVersion
 				|| !ControlConnection.termsMatchesMod(query.terms, mod)
@@ -563,26 +554,8 @@ export default class ControlConnection extends BaseConnection {
 		};
 	}
 
-	async handleModDownloadRequest(request: lib.ModDownloadRequest) {
-		let { name, version } = request;
-		let filename = `${name}_${version}.zip`;
-		let mod = this._controller.mods.get(filename);
-		if (!mod) {
-			throw new lib.RequestError(`Mod ${filename} does not exist`);
-		}
-		let modPath = path.join(this._controller.config.get("controller.mods_directory"), mod.filename);
-
-		let stream = await routes.createProxyStream(this._controller.app);
-		stream.filename = mod.filename;
-		stream.source = fs.createReadStream(modPath);
-		stream.mime = "application/zip";
-		stream.size = mod.size;
-
-		return stream.id;
-	}
-
 	async handleModDeleteRequest(request: lib.ModDeleteRequest) {
-		await this._controller.deleteMod(request.name, request.version);
+		await this._controller.modStore.deleteMod(request.name, request.version);
 	}
 
 	async handleLogSetSubscriptionsRequest(request: lib.LogSetSubscriptionsRequest) {

packages/controller/src/Controller.ts

@@ -36,6 +36,8 @@ const logSizeGauge = new Gauge(
 	"Size of all log files currently stored on the controller."
 );
 
+type InstanceId = { instanceId: number };
+
 /**
  * Manages all controller related operations
  * @alias module:controller/src/Controller
@@ -100,13 +102,13 @@ export default class Controller {
 
 		let modsDirectory = config.get("controller.mods_directory");
 		await fs.ensureDir(modsDirectory);
-		const mods = await Controller.loadModInfos(modsDirectory);
+		const modStore = await lib.ModStore.fromDirectory(modsDirectory);
 
 		return [
 			hosts,
 			instances,
 			modPacks,
-			mods,
+			modStore,
 			userManager,
 		] as const;
 	}
@@ -123,8 +125,8 @@ export default class Controller {
 		public instances = new Map<number, InstanceInfo>(),
 		/** Mapping of mod pack id to mod pack */
 		public modPacks = new Map<number, lib.ModPack>(),
-		/** Mapping of mod names to mod infos */
-		public mods = new Map<string, lib.ModInfo>(),
+		/** Mods stored on the controller */
+		public modStore = new lib.ModStore(config.get("controller.mods_directory"), new Map()),
 		/** User and roles manager for the cluster */
 		public userManager = new UserManager(config),
 	) {
@@ -140,6 +142,10 @@ export default class Controller {
 		this.wsServer = new WsServer(this);
 		this.subscriptions = new lib.SubscriptionController(this);
 
+		this.modStore.on("change", mod => {
+			this.modUpdated(mod);
+		});
+
 		// Handle subscriptions for all internal properties
 		this.subscriptions.handle(lib.HostUpdateEvent);
 		this.subscriptions.handle(lib.InstanceDetailsUpdateEvent);
@@ -509,40 +515,6 @@ export default class Controller {
 		await lib.safeOutputFile(filePath, JSON.stringify([...modPacks.values()], null, "\t"));
 	}
 
-	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()) {
-				logger.warn(`Ignoring ${entry.name}${path.sep} in mods directory.`);
-				continue;
-			}
-
-			if (entry.name.toLowerCase().endsWith(".tmp.zip")) {
-				continue;
-			}
-
-			logger.info(`Loading info for Mod ${entry.name}`);
-			let modInfo;
-			try {
-				modInfo = await lib.ModInfo.fromModFile(path.join(modsDirectory, entry.name));
-			} catch (err: any) {
-				logger.error(`Error loading mod ${entry.name}: ${err.message}`);
-				continue;
-			}
-
-			if (modInfo.filename !== entry.name) {
-				logger.error(
-					`Mod ${entry.name}'s filename does not match the expected ` +
-					`name ${modInfo.filename}, it will be ignored.`
-				);
-				continue;
-			}
-
-			mods.set(modInfo.filename, modInfo);
-		}
-		return mods;
-	}
-
 	static async loadJsonObject(filePath: string, throwOnMissing: boolean = false): Promise<any> {
 		let manifest = {};
 		try {
@@ -555,20 +527,6 @@ export default class Controller {
 		return manifest;
 	}
 
-	async deleteMod(name: string, version: string): Promise<void> {
-		let filename = `${name}_${version}.zip`;
-		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);
-		mod.isDeleted = true;
-		this.modUpdated(mod);
-	}
-
 	/**
 	 * Query controller log
 	 *
@@ -1207,17 +1165,19 @@ export default class Controller {
 		}
 	}
 
-	async sendToHostByInstanceId(requestOrEvent: any) {
+	async sendToHostByInstanceId<Req extends InstanceId, Res>(request: Req & lib.Request<Req, Res>): Promise<Res>;
+	sendToHostByInstanceId<T extends InstanceId>(event: T & lib.Event<T>): void;
+	async sendToHostByInstanceId(requestOrEvent: lib.Request<unknown, unknown> | lib.Event<unknown>) {
 		if (requestOrEvent.constructor.type === "request") {
-			return this.sendRequestToHostByInstanceId(requestOrEvent);
+			return this.sendRequestToHostByInstanceId(requestOrEvent as any);
 		}
 		if (requestOrEvent.constructor.type === "event") {
-			return this.sendEventToHostByInstanceId(requestOrEvent);
+			return this.sendEventToHostByInstanceId(requestOrEvent as any);
 		}
-		throw Error(`Unknown type ${requestOrEvent.constructor.type}.`);
+		throw Error(`Unknown type ${(requestOrEvent.constructor as any).type}.`);
 	}
 
-	async sendRequestToHostByInstanceId(request: any) {
+	async sendRequestToHostByInstanceId<Req extends InstanceId, Res>(request: Req & lib.Request<Req, Res>) {
 		let instance = this.getRequestInstance(request.instanceId);
 		let hostId = instance.config.get("instance.assigned_host");
 		if (hostId === null) {
@@ -1230,7 +1190,7 @@ export default class Controller {
 		return await connection.send(request);
 	}
 
-	sendEventToHostByInstanceId(event: any) {
+	sendEventToHostByInstanceId<T extends { instanceId: number }>(event: T & lib.Event<T>) {
 		let instance = this.getRequestInstance(event.instanceId);
 		let hostId = instance.config.get("instance.assigned_host");
 		if (hostId === null) {

packages/controller/src/HostConnection.ts

@@ -25,7 +25,6 @@ export interface HostInfo {
  * @alias module:controller/src/HostConnection
  */
 export default class HostConnection extends BaseConnection {
-	declare connector: WsServerConnector;
 	private _agent: any;
 	private _id: any;
 	private _name: any;

packages/controller/src/routes.ts

@@ -16,6 +16,17 @@ const { logger } = lib;
 
 const finished = util.promisify(nodeStream.finished);
 
+declare global {
+	// eslint-disable-next-line @typescript-eslint/no-namespace
+	namespace Express {
+		export interface Locals {
+			controller: Controller,
+			mainBundle: string,
+			devPlugins: Map<string, number>,
+			streams: Map<string, ProxyStream>
+		}
+	}
+}
 
 // Merges samples from sourceResult to destinationResult
 function mergeSamples(destinationResult: lib.CollectorResultSerialized, sourceResult: lib.CollectorResultSerialized) {
@@ -95,14 +106,14 @@ async function getMetrics(req: Request, res: Response, next: any) {
 
 function getPlugins(req: Request, res: Response) {
 	let plugins: lib.PluginWebApi[] = [];
-	for (let pluginInfo of (req.app.locals.controller as Controller).pluginInfos) {
+	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 enabled = loaded && req.app.locals.controller.config.group(pluginInfo.name).get("load_plugin") as boolean;
 		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)];
+			let stats = res.locals.webpack.devMiddleware.stats.stats[devPlugins.get(name)!];
 			web.main = stats.toJson().assetsByChunkName[name];
 		} else if (pluginInfo.manifest) {
 			web.main = pluginInfo.manifest[`${pluginInfo.name}.js`];
@@ -137,7 +148,7 @@ function validateHostToken(req: Request, res: Response, next: any) {
 		if (typeof tokenPayload === "string") {
 			throw new Error("unexpected JsonWebToken type");
 		}
-		let host = (req.app.locals.controller as Controller).hosts.get(tokenPayload.host);
+		let host = req.app.locals.controller.hosts.get(tokenPayload.host);
 		if (!host) {
 			throw new Error("invalid host");
 		}
@@ -251,12 +262,12 @@ async function uploadExport(req: Request, res: Response) {
 }
 
 
-interface ProxyStream {
+export interface ProxyStream {
 	id: string;
 	source: nodeStream.Readable | null;
 	flowing: boolean;
-	size: number | null;
-	mime: string | null;
+	size?: string;
+	mime?: string;
 	filename: string | null;
 	events: events.EventEmitter;
 	timeout: ReturnType<typeof setTimeout>;
@@ -269,8 +280,8 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
 		id,
 		source: null,
 		flowing: false,
-		size: null,
-		mime: null,
+		size: undefined,
+		mime: undefined,
 		filename: null,
 		events: new events.EventEmitter(),
 		timeout: setTimeout(() => {
@@ -289,7 +300,7 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
 }
 
 async function putStream(req: Request, res: Response) {
-	let stream = req.app.locals.streams.get(req.params.id);
+	const stream = req.app.locals.streams.get(req.params.id);
 	if (!stream || stream.source) {
 		res.sendStatus(404);
 		return;
@@ -309,35 +320,35 @@ async function putStream(req: Request, res: Response) {
 	});
 }
 
+function startStream(res: Response, stream: ProxyStream) {
+	res.append("Content-Type", stream.mime);
+	if (stream.size) {
+		res.append("Content-Length", stream.size);
+	}
+	if (stream.filename) {
+		res.append("Content-Disposition", `attachment; filename="${stream.filename}"`);
+	} else {
+		res.append("Content-Disposition", "attachment");
+	}
+	stream.source!.pipe(res);
+	stream.flowing = true;
+	res.on("close", () => {
+		stream.events.emit("close");
+	});
+	clearTimeout(stream.timeout);
+}
+
 async function getStream(req: Request, res: Response) {
-	let stream = req.app.locals.streams.get(req.params.id);
+	const stream = req.app.locals.streams.get(req.params.id);
 	if (!stream || stream.flowing) {
 		res.sendStatus(404);
 		return;
 	}
 
-	function startStream() {
-		res.append("Content-Type", stream.mime);
-		if (stream.size) {
-			res.append("Content-Length", stream.size);
-		}
-		if (stream.filename) {
-			res.append("Content-Disposition", `attachment; filename="${stream.filename}"`);
-		} else {
-			res.append("Content-Disposition", "attachment");
-		}
-		stream.source.pipe(res);
-		stream.flowing = true;
-		res.on("close", () => {
-			stream.events.emit("close");
-		});
-		clearTimeout(stream.timeout);
-	}
-
 	if (stream.source) {
-		startStream();
+		startStream(res, stream);
 	} else {
-		stream.events.on("source", startStream);
+		stream.events.on("source", () => startStream(res, stream));
 		stream.events.on("timeout", () => {
 			res.sendStatus(500);
 		});
@@ -377,7 +388,7 @@ async function uploadSave(req: Request, res: Response) {
 		let proxyStream = await createProxyStream(req.app);
 		proxyStream.source = stream;
 		proxyStream.mime = streamMime;
-		let timeout = new Promise((_, reject) => {
+		let timeout = new Promise<never>((_, reject) => {
 			proxyStream.events.on("timeout", () => {
 				stream.resume();
 				reject(new Error("Timed out establishing stream to host"));
@@ -548,9 +559,9 @@ async function uploadMod(req: Request, res: Response) {
 			await finished(writeStream);
 
 			const modInfo = await lib.ModInfo.fromModFile(path.join(modsDirectory, tempFilename));
+			checkModName(modInfo.filename);
 			await fs.rename(path.join(modsDirectory, tempFilename), path.join(modsDirectory, modInfo.filename));
-			req.app.locals.controller.mods.set(modInfo.filename, modInfo);
-			req.app.locals.controller.modUpdated(modInfo);
+			req.app.locals.controller.modStore.addMod(modInfo);
 			mods.push(modInfo.toJSON());
 
 		} catch (err: any) {

packages/host/host.ts

@@ -203,7 +203,14 @@ async function startHost() {
 	}
 
 	let hostConnector = new HostConnector(hostConfig, tlsCa, pluginInfos);
-	let host = new Host(hostConnector, args.config, hostConfig, tlsCa, pluginInfos);
+	let host = new Host(
+		hostConnector,
+		args.config,
+		hostConfig,
+		tlsCa,
+		pluginInfos,
+		...await Host.bootstrap(hostConfig)
+	);
 
 	// Handle interrupts
 	let secondSigint = false;

packages/host/src/Host.ts

@@ -263,12 +263,21 @@ export default class Host extends lib.Link {
 	_disconnecting = false;
 	_shuttingDown = false;
 
+	static async bootstrap(hostConfig: lib.HostConfig) {
+		const modsDirectory = hostConfig.get("host.mods_directory");
+		await fs.ensureDir(modsDirectory);
+		return [
+			await lib.ModStore.fromDirectory(modsDirectory),
+		] as const;
+	}
+
 	constructor(
 		connector: HostConnector,
 		hostConfigPath: string,
 		hostConfig: lib.HostConfig,
 		tlsCa: string | undefined,
-		pluginInfos: lib.PluginNodeEnvInfo[]
+		pluginInfos: lib.PluginNodeEnvInfo[],
+		public modStore: lib.ModStore,
 	) {
 		super(connector);
 		this.tlsCa = tlsCa;
@@ -478,6 +487,77 @@ export default class Host extends lib.Link {
 		}
 	}
 
+	async downloadMod(mod: lib.ModRecord) {
+		let streamId = await this.send(new lib.ModDownloadRequest(mod.name, mod.version, mod.sha1));
+
+		let url = new URL(this.config.get("host.controller_url"));
+		url.pathname += `api/stream/${streamId}`;
+		let response = await phin({
+			url, method: "GET",
+			core: { ca: this.tlsCa } as object,
+			stream: true,
+		});
+
+		const file = lib.ModInfo.filename(mod.name, mod.version);
+		const filePath = path.join(this.modStore.modsDirectory, file);
+		const tempFilePath = filePath.replace(/(\.zip)?$/, ".tmp.zip");
+		const writeStream = fs.createWriteStream(tempFilePath, { flags: "w" });
+		await events.once(writeStream, "open");
+		response.pipe(writeStream);
+		await finished(writeStream);
+		await fs.rename(tempFilePath, filePath);
+
+		const modInfo = await this.modStore.loadFile(file);
+		if (mod.sha1 && mod.sha1 !== modInfo.sha1) {
+			throw new Error(`Checksum mismatch downloading ${file} from controller`);
+		}
+		logger.info(`Downloaded ${file} from controller`);
+		return modInfo;
+	}
+
+	private _fetchModQueue = new lib.AsyncSerialCallback(this._fetchModInternal.bind(this));
+
+	async fetchMod(mod: lib.ModRecord) {
+		return await this._fetchModQueue.invoke(mod);
+	}
+
+	private async _fetchModInternal(mod: lib.ModRecord) {
+		// The mod may have already been downloaded in a previous call
+		let modInfo = this.modStore.getMod(mod.name, mod.version);
+		if (modInfo && (!mod.sha1 || mod.sha1 === modInfo.sha1)) {
+			return modInfo;
+		}
+
+		// The mods folder may be shared with the controller and thus
+		// already have this mod, try loading it.
+		try {
+			modInfo = await this.modStore.loadMod(mod.name, mod.version);
+		} catch (err: any) {
+			if (err.code !== "ENOENT") {
+				throw err;
+			}
+		}
+		if (modInfo && (!mod.sha1 || mod.sha1 === modInfo.sha1)) {
+			return modInfo;
+		}
+
+		// Fetch the mod from the controller.
+		return await this.downloadMod(mod);
+	}
+
+	async fetchMods(mods: Iterable<lib.ModRecord>) {
+		const modInfos: Promise<lib.ModInfo>[] = [];
+		for (const mod of mods) {
+			// XXX These mods ship with Factorio, but the list of mods may
+			// change and should not be hardcoded here.
+			if (["core", "base"].includes(mod.name)) {
+				continue;
+			}
+			modInfos.push(this.fetchMod(mod));
+		}
+		return await Promise.all(modInfos);
+	}
+
 	async handleInstanceAssignInternalRequest(request: lib.InstanceAssignInternalRequest) {
 		let { instanceId, config } = request;
 		let instanceInfo = this.instanceInfos.get(instanceId);

packages/host/src/Instance.ts

@@ -634,6 +634,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 	 * Windows hard links are used instead due to symlinks being privileged.
 	 */
 	async syncMods() {
+		this.logger.info("Syncing mods");
 		const modPackId = this.config.get("factorio.mod_pack");
 		let modPack;
 		if (modPackId === null) {
@@ -645,9 +646,7 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		// TODO validate factorioVersion
 
-		if (!this._host.config.get("host.mods_directory_is_shared")) {
-			throw new Error("Fetching mods is not implemented");
-		}
+		const mods = await this._host.fetchMods(modPack.mods.values());
 
 		await fs.ensureDir(this.path("mods"));
 
@@ -667,8 +666,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		// Add mods from mod the pack
 		const modsDir = this._host.config.get("host.mods_directory");
-		for (let mod of this.activeModPack.mods.values()) {
-			const modFile = `${mod.name}_${mod.version}.zip`;
+		for (let mod of mods) {
+			const modFile = mod.filename;
 			const target = path.join(modsDir, modFile);
 			const link = this.path("mods", modFile);
 
@@ -994,7 +993,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 		try {
 			await this.prepare();
 			saveName = await this.prepareSave(saveName);
-		} catch (err) {
+		} catch (err: any) {
+			this.logger.error(`Error preparing instance: ${err.message}`);
 			this.notifyExit();
 			await this.sendSaveListUpdate();
 			throw err;
@@ -1002,7 +1002,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		try {
 			await this.start(saveName);
-		} catch (err) {
+		} catch (err: any) {
+			this.logger.error(`Error starting ${saveName}: ${err.message}`);
 			await this.stop();
 			throw err;
 		}
@@ -1016,7 +1017,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		try {
 			await this.prepare();
-		} catch (err) {
+		} catch (err: any) {
+			this.logger.error(`Error preparing instance: ${err.message}`);
 			this.notifyExit();
 			await this.sendSaveListUpdate();
 			throw err;
@@ -1025,7 +1027,8 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 		let { scenario, seed, mapGenSettings, mapSettings } = request;
 		try {
 			await this.startScenario(scenario, seed, mapGenSettings, mapSettings);
-		} catch (err) {
+		} catch (err: any) {
+			this.logger.error(`Error starting scenario ${scenario}: ${err.message}`);
 			await this.stop();
 			throw err;
 		}
@@ -1040,11 +1043,10 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 		try {
 			this.logger.verbose("Writing server-settings.json");
 			await this.writeServerSettings();
-
-			this.logger.verbose("Creating save .....");
 			await this.syncMods();
 
-		} catch (err) {
+		} catch (err: any) {
+			this.logger.error(`Error preparing instance: ${err.message}`);
 			this.notifyExit();
 			await this.sendSaveListUpdate();
 			throw err;
@@ -1052,7 +1054,14 @@ rcon.print(game.table_to_json(players))`.replace(/\r?\n/g, " ");
 
 		this.server.on("exit", () => this.notifyExit());
 		let { name, seed, mapGenSettings, mapSettings } = request;
-		await this.server.create(name, seed, mapGenSettings, mapSettings);
+
+		try {
+			this.logger.info("Creating save");
+			await this.server.create(name, seed, mapGenSettings, mapSettings);
+		} catch (err: any) {
+			this.logger.error(`Error creating save ${name}: ${err.message}`);
+			throw err;
+		}
 		await this.sendSaveListUpdate();
 		this.logger.info("Successfully created save");
 	}

packages/lib/build_mod.js

@@ -12,10 +12,10 @@ const yargs = require("yargs");
 const finished = util.promisify(stream.finished);
 
 
-async function buildMod(args, info) {
+async function buildMod(args, info, modName) {
 	if (args.build) {
 		await fs.ensureDir(args.outputDir);
-		let modName = `${info.name}_${info.version}`;
+		modName = args.modName ?? `${info.name}_${info.version}`;
 
 		if (args.pack) {
 			let zip = new JSZip();

packages/lib/index.ts

@@ -29,4 +29,5 @@ export * from "./src/zip_ops";
 export * from "./src/subscriptions";
 
 export { default as ExponentialBackoff } from "./src/ExponentialBackoff";
+export { default as ModStore } from "./src/ModStore";
 export { default as RateLimiter } from "./src/RateLimiter";

packages/lib/src/ModStore.ts

@@ -0,0 +1,110 @@
+import fs from "fs-extra";
+import path from "path";
+import { logger } from "./logging";
+import { ModInfo } from "./data";
+import EventEmitter from "events";
+
+abstract class TypedEventEmitter<
+	Keys extends string,
+	Events extends Record<Keys, (...args: [...any]) => void>,
+> extends EventEmitter {
+	declare ["on"]: <E extends Keys>(event: E, listener: Events[E]) => this;
+	declare ["off"]: <E extends Keys>(event: E, listener: Events[E]) => this;
+	declare ["once"]: <E extends Keys>(event: E, listener: Events[E]) => this;
+	declare ["emit"]: <E extends Keys>(event: E, ...args: Parameters<Events[E]>) => boolean;
+}
+
+export interface ModStoreEvents {
+	/** A stored mod was created, updated or deleted */
+	change: (mod: ModInfo) => void;
+}
+
+export default class ModStore extends TypedEventEmitter<keyof ModStoreEvents, ModStoreEvents> {
+	constructor(
+		public modsDirectory: string,
+		public files: Map<string, ModInfo>,
+	) {
+		super();
+	}
+
+	getMod(name: string, version: string, sha1?: string) {
+		const mod = this.files.get(ModInfo.filename(name, version));
+		if (!mod || sha1 && sha1 !== mod.sha1) {
+			return undefined;
+		}
+		return mod;
+	}
+
+	hasMod(name: string, version: string, sha1?: string): boolean {
+		const mod = this.files.get(ModInfo.filename(name, version));
+		return Boolean(mod && (!sha1 || sha1 !== mod.sha1));
+	}
+
+	mods() { return this.files.values(); }
+
+	static async loadFile(modsDirectory: string, file: string) {
+		let modInfo = await ModInfo.fromModFile(path.join(modsDirectory, file));
+		if (modInfo.filename !== file) {
+			throw new Error(`filename does not match the expected name ${modInfo.filename}.`);
+		}
+		return modInfo;
+	}
+
+	async loadFile(file: string) {
+		let modInfo = await ModStore.loadFile(this.modsDirectory, file);
+		this.files.set(modInfo.filename, modInfo);
+		this.emit("change", modInfo);
+		return modInfo;
+	}
+
+	async deleteFile(file: string) {
+		let modInfo = this.files.get(file);
+		if (!modInfo) {
+			throw new Error(`Mod ${file} does not exist`);
+		}
+
+		await fs.unlink(path.join(this.modsDirectory, file));
+		this.files.delete(file);
+		modInfo.isDeleted = true;
+		this.emit("change", modInfo);
+	}
+
+	addMod(modInfo: ModInfo) {
+		this.files.set(modInfo.filename, modInfo);
+		this.emit("change", modInfo);
+	}
+
+	async loadMod(name: string, version: string) {
+		return await this.loadFile(ModInfo.filename(name, version));
+	}
+
+	async deleteMod(name: string, version: string) {
+		return await this.deleteFile(ModInfo.filename(name, version));
+	}
+
+	static async fromDirectory(modsDirectory: string) {
+		const files = new Map<string, ModInfo>();
+		for (let entry of await fs.readdir(modsDirectory, { withFileTypes: true })) {
+			if (entry.isDirectory()) {
+				logger.warn(`Ignoring ${entry.name}${path.sep} in mods directory.`);
+				continue;
+			}
+
+			if (entry.name.toLowerCase().endsWith(".tmp.zip")) {
+				continue;
+			}
+
+			logger.info(`Loading info for Mod ${entry.name}`);
+			let modInfo: ModInfo;
+			try {
+				modInfo = await this.loadFile(modsDirectory, entry.name);
+			} catch (err: any) {
+				logger.error(`Error loading mod ${entry.name}: ${err.message}`);
+				continue;
+			}
+			files.set(modInfo.filename, modInfo);
+		}
+
+		return new this(modsDirectory, files);
+	}
+}

packages/lib/src/config/definitions.ts

@@ -21,6 +21,7 @@ ControllerGroup.define({
 	name: "mods_directory",
 	title: "Mods Directory",
 	description: "Path to directory where mods shared with the cluster are stored.",
+	restartRequired: true,
 	type: "string",
 	initial_value: "mods",
 });
@@ -202,16 +203,10 @@ HostGroup.define({
 	name: "mods_directory",
 	title: "Mods Directory",
 	description: "Path to directory where mods for instances are cached.",
+	restartRequired: true,
 	type: "string",
 	initial_value: "mods",
 });
-HostGroup.define({
-	name: "mods_directory_is_shared",
-	title: "Mods Directory is Shared",
-	description: "True if the mods directory is shared with the controller.",
-	type: "boolean",
-	initial_value: true,
-});
 HostGroup.define({
 	name: "instances_directory",
 	description: "Path to directory to store instances in.",
@@ -264,7 +259,6 @@ export class HostConfig extends classes.Config {
 	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;

packages/lib/src/data/ModInfo.ts

@@ -90,7 +90,17 @@ export default class ModInfo {
 	 * Expected name of zip file containing this mod.
 	 */
 	get filename() {
-		return `${this.name}_${this.version}.zip`;
+		return ModInfo.filename(this.name, this.version);
+	}
+
+	/**
+	 * Expected name of zip file containing mod with given name and version.
+	 * @param name - Mod's name
+	 * @param version - Mod's version
+	 * @returns string containing {name}_{version}.zip
+	 */
+	static filename(name: string, version: string) {
+		return `${name}_${version}.zip`;
 	}
 
 	/**

packages/lib/src/data/messages_mod.ts

@@ -114,15 +114,17 @@ export class ModGetRequest {
 	constructor(
 		public name: string,
 		public version: string,
+		public sha1?: string,
 	) { }
 
 	static jsonSchema = Type.Object({
 		"name": Type.String(),
 		"version": Type.String(),
+		"sha1": Type.Optional(Type.String()),
 	});
 
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.name, json.version);
+		return new this(json.name, json.version, json.sha1);
 	}
 
 	static Response = ModInfo;
@@ -198,22 +200,24 @@ export class ModSearchRequest {
 export class ModDownloadRequest {
 	declare ["constructor"]: typeof ModDownloadRequest;
 	static type = "request" as const;
-	static src = "control" as const;
+	static src = ["host", "control"] as const;
 	static dst = "controller" as const;
 	static permission = "core.mod.download" as const;
 
 	constructor(
 		public name: string,
 		public version: string,
+		public sha1?: string,
 	) { }
 
 	static jsonSchema = Type.Object({
 		"name": Type.String(),
 		"version": Type.String(),
+		"sha1": Type.Optional(Type.String()),
 	});
 
 	static fromJSON(json: Static<typeof this.jsonSchema>) {
-		return new this(json.name, json.version);
+		return new this(json.name, json.version, json.sha1);
 	}
 
 	static Response = JsonString;

packages/lib/src/helpers.ts

@@ -107,6 +107,54 @@ export class AsyncSerialMergingCallback {
 	}
 }
 
+/**
+ * Helper to serialise calls to an async callback.
+ *
+ * Provides a convenient interface to serialise the calls made to an async
+ * callback such that no overlapping invocations of the callback is made,
+ * and successive calls are served on a first in first out basis.  If the
+ * callback is already running when another invocation is requested the
+ * request will be queued until the callback returns and then the callback
+ * called again.
+ */
+export class AsyncSerialCallback<
+	Callback extends (...args: Parameters<Callback>) => Promise<Awaited<ReturnType<Callback>>>,
+> {
+	private _currentlyRunning = false;
+	private _currentlyWaiting: (() => void)[] = [];
+
+	/**
+	 * @param callback - Async function to serialise access to
+	 */
+	constructor(
+		public callback: Callback,
+	) { }
+
+	/**
+	 * Invoke the assosiated callback.
+	 *
+	 * If the callback is currently running then this will wait until the
+	 * existing invocation finishes and then invoke it again. If called
+	 * multiple times while an invocation is running the calls will be
+	 * merged into one call.
+	 */
+	async invoke(...args: Parameters<Callback>) {
+		if (this._currentlyRunning) {
+			await new Promise<void>(resolve => {
+				this._currentlyWaiting.push(resolve);
+			});
+		}
+
+		this._currentlyRunning = true;
+		try {
+			return await this.callback(...args);
+		} finally {
+			this._currentlyRunning = false;
+			this._currentlyWaiting.shift()?.();
+		}
+	}
+}
+
 
 /**
  * Read stream to the end and return its content

packages/web_ui/src/components/ModsPage.tsx

@@ -132,7 +132,7 @@ export default function ModsPage() {
 	let [modList] = useModList();
 	let [modPackList] = useModPackList();
 
-	function actions(mod: lib.ModInfo|lib.ModRecord) {
+	function actions(mod: lib.ModInfo) {
 		return <Space>
 			{account.hasPermission("core.mod.download")
 				&& <Typography.Link
@@ -151,7 +151,7 @@ export default function ModsPage() {
 			}
 			{account.hasPermission("core.mod.delete")
 				&& <Popconfirm
-					title={`Delete ${mod.name}_${mod.version}.zip?`}
+					title={`Delete ${mod.filename}?`}
 					onConfirm={event => {
 						control.send(
 							new lib.ModDeleteRequest(mod.name, mod.version)

test/file/path_traversing_mod/info.json

@@ -0,0 +1,9 @@
+{
+    "name": "../bad_path_mod",
+    "version": "1.0.0",
+    "title": "Path Traversing Mod",
+    "author": "Me",
+    "description": "A mod who's name traverses the filesystem",
+    "factorio_version": "1.1",
+    "dependencies": []
+}

test/integration/clusterio.js

@@ -32,9 +32,12 @@ async function startAltHost() {
 	let configPath = path.join("temp", "test", config);
 	await fs.remove(configPath);
 	await fs.remove(path.join("temp", "test", "alt-instances"));
+	await fs.remove(path.join("temp", "test", "alt-mods"));
 	await execCtl(`host create-config --id 5 --name alt-host --generate-token --output ${config}`);
 	await exec(`node ../../packages/host --config ${config} config set host.tls_ca ../../test/file/tls/cert.pem`);
 	await exec(`node ../../packages/host --config ${config} config set host.instances_directory alt-instances`);
+	await exec(`node ../../packages/host --config ${config} config set host.factorio_directory ../../factorio`);
+	await exec(`node ../../packages/host --config ${config} config set host.mods_directory alt-mods`);
 	return await spawn("alt-host:", `node ../../packages/host run --config ${config}`, /Started host/);
 }
 
@@ -176,6 +179,23 @@ describe("Integration of Clusterio", function() {
 				assert(hosts.has(5), "Host list was not updated");
 			});
 		});
+		it("should download mods from controller", async function() {
+			// On windows there's currently no way to automate graceful shutdown of the host
+			// process as CTRL+C is some weird terminal thing and SIGINT isn't a thing.
+			if (process.platform === "win32") {
+				this.skip();
+			}
+
+			slowTest(this);
+			await runWithAltHost(async () => {
+				const clusterioLib = path.join("temp", "test", "alt-mods", "clusterio_lib_0.1.2.zip");
+				assert(!await fs.pathExists(clusterioLib), "mod was downloaded before the test");
+				await execCtl("instance create alt-mod --id 98");
+				await execCtl("instance assign alt-mod 5");
+				await execCtl("instance save create alt-mod");
+				assert(await fs.pathExists(clusterioLib), "mod was not downloaded by the test");
+			});
+		});
 	});
 
 	describe("clusterioctl", function() {
@@ -934,6 +954,21 @@ describe("Integration of Clusterio", function() {
 					"mod not present in mods directory"
 				);
 			});
+			it("rejects path traversing mods", async function() {
+				await libBuildMod.build({
+					build: true,
+					pack: true,
+					sourceDir: path.join("test", "file", "path_traversing_mod"),
+					outputDir: path.join("temp", "test"),
+					modName: "path_traversing_mod_1.0.0",
+				});
+				await fs.remove(path.join("temp", "test", "bad_path_mod_1.0.0.zip"));
+				await assert.rejects(execCtl("mod upload path_traversing_mod_1.0.0.zip"));
+				assert(
+					!await fs.pathExists(path.join("temp", "test", "bad_path_mod_1.0.0.zip")),
+					"mod executed a path traversal attack",
+				);
+			});
 		});
 
 		describe("mod show", function() {

test/integration/index.js

@@ -18,6 +18,8 @@ require("../../plugins/research_sync/dist/plugin/info");
 require("../../plugins/statistics_exporter/dist/plugin/info");
 require("../../plugins/subspace_storage/dist/plugin/info");
 
+const subspaceStorageVersion = "1.99.8";
+const subspaceStorageGitHubTag = "1.99.6"; // Tag subspaceStorageVersion was released under on GitHub
 
 class TestControl extends lib.Link {
 	constructor(connector, subscribe = true) {
@@ -131,6 +133,7 @@ let control;
 let url = "https://localhost:4443/";
 let controlToken = jwt.sign({ aud: "user", user: "test" }, Buffer.from("TestSecretDoNotUse", "base64"));
 let instancesDir = path.join("temp", "test", "instances");
+let modsDir = path.join("temp", "test", "mods");
 let databaseDir = path.join("temp", "test", "database");
 let pluginListPath = path.join("temp", "test", "plugin-list.json");
 let controllerConfigPath = path.join("temp", "test", "config-controller.json");
@@ -186,8 +189,9 @@ before(async function() {
 		format: new lib.TerminalFormat(),
 	}));
 
-	await fs.remove(databaseDir);
 	await fs.remove(instancesDir);
+	await fs.remove(modsDir);
+	await fs.remove(databaseDir);
 
 	await fs.remove(pluginListPath);
 	await fs.remove(controllerConfigPath);
@@ -196,6 +200,32 @@ before(async function() {
 
 	await fs.ensureDir(path.join("temp", "test"));
 
+	await fs.ensureDir(modsDir);
+	await lib.build({
+		build: true,
+		pack: true,
+		sourceDir: "packages/host/lua/clusterio_lib",
+		outputDir: modsDir,
+	});
+	const ssv = subspaceStorageVersion;
+	const ssTag = subspaceStorageGitHubTag;
+	const ssZip = `subspace_storage_${ssv}.zip`;
+	if (!await fs.pathExists(`mods/${ssZip}`)) {
+		console.log(`Downloading ${ssZip}`);
+		const ssUrl = `https://github.com/clusterio/subspace_storage/releases/download/${ssTag}/${ssZip}`;
+		let response = await phin({
+			url: ssUrl,
+			method: "GET",
+			stream: false, // phin leaks open sockets when stream and followRedirects is combined
+			followRedirects: true,
+		});
+		if (response.statusCode !== 200) {
+			throw new Error(`${ssUrl} returned ${response.statusCode} ${response.statusMessage}`);
+		}
+		await fs.outputFile(`mods/${ssZip}`, response.body);
+	}
+	await fs.copyFile(`mods/${ssZip}`, `${modsDir}/${ssZip}`);
+
 	await exec("node ../../packages/controller config set controller.auth_secret TestSecretDoNotUse");
 	await exec("node ../../packages/controller config set controller.http_port 8880");
 	await exec("node ../../packages/controller config set controller.https_port 4443");
@@ -233,7 +263,7 @@ before(async function() {
 	testPack.name = "subspace_storage-pack";
 	testPack.factorioVersion = "1.1.0";
 	testPack.mods.set("clusterio_lib", { name: "clusterio_lib", enabled: true, version: "0.1.2" });
-	testPack.mods.set("subspace_storage", { name: "subspace_storage", enabled: true, version: "1.99.8" });
+	testPack.mods.set("subspace_storage", { name: "subspace_storage", enabled: true, version: subspaceStorageVersion });
 	await control.sendTo("controller", new lib.ModPackCreateRequest(testPack));
 	await control.sendTo(
 		"controller",
@@ -279,6 +309,7 @@ module.exports = {
 	url,
 	controlToken,
 	instancesDir,
+	modsDir,
 	databaseDir,
 	controllerConfigPath,
 	hostConfigPath,

test/lib/helpers.js

@@ -104,6 +104,94 @@ describe("lib/helpers", function() {
 		});
 	});
 
+	describe("class AsyncSerialCallback", function() {
+		let started;
+		let ended;
+		let serialiser;
+		beforeEach(function() {
+			started = 0;
+			ended = 0;
+			serialiser = new lib.AsyncSerialCallback(
+				(input) => new Promise((resolve, reject) => {
+					started += 1;
+					process.nextTick(() => {
+						ended += 1;
+						if (input !== 99) {
+							resolve(input);
+						} else {
+							reject(new Error("99"));
+						}
+					});
+				})
+			);
+		});
+		it("should await the callback when invoked", async function() {
+			await serialiser.invoke();
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 1, "incorrect ended count");
+		});
+		it("should return the result of the callback", async function() {
+			const id = await serialiser.invoke(42);
+			assert.equal(id, 42, "returned result not passed");
+		});
+		it("should immediately call the callback", async function() {
+			let promise = serialiser.invoke();
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 0, "incorrect ended count");
+			await promise;
+			assert.equal(started, 1, "incorrect started count");
+			assert.equal(ended, 1, "incorrect ended count");
+		});
+		it("should call the callback twice if invoked in twice in paralell", async function() {
+			await Promise.all([
+				serialiser.invoke(),
+				serialiser.invoke(),
+			]);
+			assert.equal(started, 2, "incorrect started count");
+			assert.equal(ended, 2, "incorrect ended count");
+		});
+		it("should call the callback many if invoked many times in paralell", async function() {
+			await Promise.all([
+				serialiser.invoke(),
+				serialiser.invoke(),
+				serialiser.invoke(),
+				serialiser.invoke(),
+			]);
+			assert.equal(started, 4, "incorrect started count");
+			assert.equal(ended, 4, "incorrect ended count");
+		});
+		it("should serialise calls", async function() {
+			let events = [];
+			let calls = await Promise.all([
+				serialiser.invoke(11).then((res) => { events.push(1); return res; }),
+				serialiser.invoke(12).then((res) => { events.push(2); return res; }),
+				serialiser.invoke(13).then((res) => { events.push(3); return res; }),
+				serialiser.invoke(14).then((res) => { events.push(4); return res; }),
+			]);
+			assert.deepEqual(events, [1, 2, 3, 4]);
+			assert.deepEqual(calls, [11, 12, 13, 14]);
+		});
+		it("should forward rejections", async function() {
+			await assert.rejects(serialiser.invoke(99), new Error("99"));
+		});
+		it("should handle queued rejections", async function() {
+			let events = [];
+			let calls = await Promise.allSettled([
+				serialiser.invoke(11).then((res) => { events.push(1); return res; }),
+				serialiser.invoke(99).then((res) => { events.push(2); return res; }),
+				serialiser.invoke(13).then((res) => { events.push(3); return res; }),
+				serialiser.invoke(14).then((res) => { events.push(4); return res; }),
+			]);
+			assert.deepEqual(events, [1, 3, 4]);
+			assert.deepEqual(calls, [
+				{ status: "fulfilled", value: 11 },
+				{ status: "rejected", reason: new Error("99") },
+				{ status: "fulfilled", value: 13 },
+				{ status: "fulfilled", value: 14 },
+			]);
+		});
+	});
+
 	function parse(input, attributes) {
 		const result = lib.parseSearchString(input, attributes);
 		if (!result.issues.length) {