Add upload save feature

Add /api/upload-save endpoint that streams the data via the slave
assigned to the target instance to the instance saves directory, along
with a ctl command and web interface button to do the upload.

packages/master/src/BaseConnection.js

@@ -23,25 +23,7 @@ class BaseConnection extends libLink.Link {
 	}
 
 	async forwardRequestToInstance(message, request) {
-		let instance = this._master.instances.get(message.data.instance_id);
-		if (!instance) {
-			throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`);
-		}
-
-		let slaveId = instance.config.get("instance.assigned_slave");
-		if (slaveId === null) {
-			throw new libErrors.RequestError("Instance is not assigned to a slave");
-		}
-
-		let connection = this._master.wsServer.slaveConnections.get(slaveId);
-		if (!connection) {
-			throw new libErrors.RequestError("Slave containing instance is not connected");
-		}
-		if (request.plugin && !connection.plugins.has(request.plugin)) {
-			throw new libErrors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`);
-		}
-
-		return await request.send(connection, message.data);
+		return await this._master.forwardRequestToInstance(request, message.data);
 	}
 
 	async forwardEventToInstance(message, event) {

packages/master/src/Master.js

@@ -2,7 +2,6 @@
 const compression = require("compression");
 const events = require("events");
 const express = require("express");
-const expressFileupload = require("express-fileupload");
 const fs = require("fs-extra");
 const http = require("http");
 const https = require("https");
@@ -49,6 +48,7 @@ class Master {
 
 		this.app = express();
 		this.app.locals.master = this;
+		this.app.locals.streams = new Map();
 
 		/**
 		 * Mapping of slave id to slave info
@@ -336,7 +336,6 @@ class Master {
 			hitCounter.inc();
 			next();
 		});
-		app.use(expressFileupload());
 		app.use(compression());
 
 		// Set folder to serve static content from (the website)
@@ -492,6 +491,41 @@ class Master {
 			});
 		};
 	}
+
+	/**
+	 * Forward the given request to the slave contaning the instance
+	 *
+	 * Finds the slave which the instance with the given instance ID is
+	 * currently assigned to and forwards the request to that instance.
+	 * The request data must have an instance_id parameter.
+	 *
+	 * @param {module:lib/link.Request} request - The request to send.
+	 * @param {object} data - Data to pass with the request.
+	 * @param {number} data.instance_id -
+	 *     ID of instance to send this request towards
+	 * @returns {object} data returned from the request.
+	 */
+	async forwardRequestToInstance(request, data) {
+		let instance = this.instances.get(data.instance_id);
+		if (!instance) {
+			throw new libErrors.RequestError(`Instance with ID ${data.instance_id} does not exist`);
+		}
+
+		let slaveId = instance.config.get("instance.assigned_slave");
+		if (slaveId === null) {
+			throw new libErrors.RequestError("Instance is not assigned to a slave");
+		}
+
+		let connection = this.wsServer.slaveConnections.get(slaveId);
+		if (!connection) {
+			throw new libErrors.RequestError("Slave containing instance is not connected");
+		}
+		if (request.plugin && !connection.plugins.has(request.plugin)) {
+			throw new libErrors.RequestError(`Slave containing instance does not have ${request.plugin} plugin`);
+		}
+
+		return await request.send(connection, data);
+	}
 }
 
 module.exports = Master;

packages/master/src/routes.js

@@ -1,13 +1,18 @@
 "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 util = require("util");
 
 const libHelpers = require("@clusterio/lib/helpers");
 const libLink = require("@clusterio/lib/link");
 const libPlugin = require("@clusterio/lib/plugin");
 const libPrometheus = require("@clusterio/lib/prometheus");
+const { logger } = require("@clusterio/lib/logging");
 
 const { endpointHitCounter } = require("./metrics");
 
@@ -106,6 +111,36 @@ function validateSlaveToken(req, res, next) {
 	next();
 }
 
+function validateUserToken(req, res, next) {
+	let token = req.header("x-access-token");
+	if (!token) {
+		res.sendStatus(401);
+		return;
+	}
+
+	try {
+		let tokenPayload = jwt.verify(
+			token,
+			req.app.locals.master.config.get("master.auth_secret"),
+			{ audience: "user" }
+		);
+		let user = req.app.locals.master.userManager.users.get(tokenPayload.user);
+		if (!user) {
+			throw new Error("invalid user");
+		}
+		if (tokenPayload.iat < user.tokenValidAfter) {
+			throw new Error("invalid token");
+		}
+		res.locals.user = user;
+
+	} catch (err) {
+		res.sendStatus(401);
+		return;
+	}
+
+	next();
+}
+
 // Handle an uploaded export package.
 async function uploadExport(req, res) {
 	endpointHitCounter.labels(req.route.path).inc();
@@ -142,6 +177,191 @@ async function uploadExport(req, res) {
 	res.sendStatus(200);
 }
 
+async function createProxyStream(app) {
+	let asyncRandomBytes = util.promisify(crypto.randomBytes);
+	let id = (await asyncRandomBytes(8)).toString("hex");
+	let stream = {
+		id,
+		flowing: false,
+		size: null,
+		mime: null,
+		events: new events.EventEmitter(),
+		timeout: setTimeout(() => {
+			stream.events.emit("timeout");
+		}, app.locals.master.config.get("master.proxy_stream_timeout") * 1000),
+	};
+	stream.events.on("close", () => {
+		clearTimeout(stream.timeout);
+		app.locals.streams.delete(id);
+	});
+	stream.events.on("timeout", () => {
+		stream.events.emit("close");
+	});
+	app.locals.streams.set(id, stream);
+	return stream;
+}
+
+async function getStream(req, res) {
+	let 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);
+		}
+		stream.source.pipe(res);
+		stream.flowing = true;
+		res.on("finish", () => {
+			stream.events.emit("close");
+		});
+		clearTimeout(stream.timeout);
+	}
+
+	if (stream.source) {
+		startStream();
+	} else {
+		stream.events.on("source", startStream);
+		stream.events.on("timeout", () => {
+			res.sendStatus(500);
+		});
+	}
+}
+
+
+async function uploadSave(req, res) {
+	try {
+		res.locals.user.checkPermission("core.instance.save.upload");
+	} catch (err) {
+		res.status(403).json({ request_errors: [err.message] });
+		return;
+	}
+
+	let contentType = req.get("Content-Type");
+	let match = /^([!#$%&'*+\-.^_`|~0-9A-Za-z]+\/[!#$%&'*+\-.^_`|~0-9A-Za-z]+)/.exec(contentType);
+	if (!match) {
+		res.status(415).json({ request_errors: ["invalid Content-Type"] });
+		return;
+	}
+	let mimeType = match[1].toLowerCase();
+
+	let tasks = [];
+	let errors = [];
+	let requestErrors = [];
+	let saves = [];
+
+	async function handleFile(instanceId, stream, filename, streamMime) {
+		let proxyStream = await createProxyStream(req.app);
+		proxyStream.source = stream;
+		proxyStream.mime = streamMime;
+		let timeout = new Promise((_, reject) => {
+			proxyStream.events.on("timeout", () => {
+				stream.resume();
+				reject(new Error("Timed out establishing stream to slave"));
+			});
+		});
+
+		try {
+			let result = await Promise.race([
+				req.app.locals.master.forwardRequestToInstance(libLink.messages.pullSave, {
+					instance_id: instanceId,
+					stream_id: proxyStream.id,
+					filename,
+				}),
+				timeout,
+			]);
+			saves.push(result.save);
+
+		} catch (err) {
+			proxyStream.events.emit("close");
+			logger.error(`Error uploading save: ${err.message}`);
+			errors.push(err.message);
+			stream.resume();
+		}
+	}
+
+	if (mimeType === "multipart/form-data") {
+		await new Promise(resolve => {
+			let fields = {};
+			let busboy = new Busboy({ headers: req.headers });
+			busboy.on("file", (fieldname, stream, filename, transferEncoding, fileMime) => {
+				if (fields.instanceId === undefined) {
+					requestErrors.push("instance_id must come before files uploaded");
+				}
+
+				if (fileMime !== "application/zip") {
+					requestErrors.push("invalid file Content-Type");
+				}
+
+				if (!filename.endsWith(".zip")) {
+					requestErrors.push("filename must end with .zip");
+				}
+
+				if (errors.length || requestErrors.length) {
+					stream.resume();
+					return;
+				}
+
+				tasks.push(handleFile(fields.instanceId, stream, filename, fileMime));
+			});
+			busboy.on("field", (fieldname, value, fieldnameTruncated, valueTruncated, transferEncoding, valueMime) => {
+				if (fieldname === "instance_id") {
+					fields.instanceId = Number.parseInt(value, 10);
+					if (Number.isNaN(fields.instanceId)) {
+						requestErrors.push("invalid instance_id");
+					}
+				}
+			});
+			busboy.on("finish", resolve);
+			busboy.on("error", (err) => {
+				logger.error(`Error parsing multipart request in upload-save:\n${err.stack}`);
+				errors.push(err.message);
+			});
+			req.pipe(busboy);
+		});
+
+	} else if (mimeType === "application/zip") {
+		let filename = req.query.filename;
+		if (typeof filename !== "string") {
+			requestErrors.push("Missing or invalid filename parameter");
+		} else if (!filename.endsWith(".zip")) {
+			requestErrors.push("filename must end with .zip");
+		}
+		let instanceId = Number.parseInt(req.query.instance_id, 10);
+		if (Number.isNaN(instanceId)) {
+			requestErrors.push("Missing or invalid instance_id parameter");
+		}
+
+		if (errors.length || requestErrors.length) {
+			req.resume();
+		} else {
+			tasks.push(handleFile(instanceId, req, filename, mimeType));
+		}
+
+	} else {
+		res.status(415).json({ request_errors: ["invalid Content-Type"] });
+		return;
+	}
+
+	await Promise.all(tasks);
+
+	if (errors.length) {
+		res.status(500);
+		res.json({ errors, request_errors: requestErrors });
+		return;
+	}
+	if (requestErrors.length) {
+		res.status(400);
+		res.json({ request_errors: requestErrors });
+		return;
+	}
+
+	res.json({ saves });
+}
+
 function addRouteHandlers(app) {
 	app.get("/metrics", (req, res, next) => getMetrics(req, res, next).catch(next));
 	app.get("/api/plugins", getPlugins);
@@ -149,6 +369,11 @@ function addRouteHandlers(app) {
 		validateSlaveToken,
 		(req, res, next) => uploadExport(req, res).catch(next)
 	);
+	app.get("/api/stream/:id", (req, res, next) => getStream(req, res).catch(next));
+	app.post("/api/upload-save",
+		validateUserToken,
+		(req, res, next) => uploadSave(req, res).catch(next)
+	);
 }
 
 // Routes used in the web interface and served by the master server
@@ -170,4 +395,5 @@ const webRoutes = [
 module.exports = {
 	addRouteHandlers,
 	webRoutes,
+	createProxyStream,
 };