Add download save feature

Add command and web interface button to download saves from instances.

packages/ctl/ctl.js

@@ -26,6 +26,7 @@ const { ConsoleTransport, levels, logger } = require("@clusterio/lib/logging");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
 const libHelpers = require("@clusterio/lib/helpers");
 const libFactorio = require("@clusterio/lib/factorio");
+const libFileOps = require("@clusterio/lib/file_ops");
 
 
 function print(...content) {
@@ -429,6 +430,51 @@ instanceCommands.add(new libCommand.Command({
 	},
 }));
 
+instanceCommands.add(new libCommand.Command({
+	definition: ["download-save <instance> <save>", "Download a save from an instance", (yargs) => {
+		yargs.positional("instance", { describe: "Instance to download save from", type: "string" });
+		yargs.positional("save", { describe: "Save to download", type: "string" });
+	}],
+	handler: async function(args, control) {
+		let instanceId = await libCommand.resolveInstance(control, args.instance);
+		let result = await libLink.messages.downloadSave.send(control, {
+			instance_id: instanceId,
+			save: args.save,
+		});
+
+		let url = new URL(control.config.get("control.master_url"));
+		url.pathname += `api/stream/${result.stream_id}`;
+		let response = await phin({
+			url, method: "GET",
+			core: { ca: control.tlsCa },
+			stream: true,
+		});
+
+		let stream;
+		let tempFilename = args.save.replace(/(\.zip)?$/, ".tmp.zip");
+		while (true) {
+			try {
+				stream = fs.createWriteStream(tempFilename, { flags: "wx" });
+				await events.once(stream, "open");
+				break;
+			} catch (err) {
+				if (err.code === "EEXIST") {
+					tempFilename = await libFileOps.findUnusedName(".", tempFilename, ".tmp.zip");
+				} else {
+					throw err;
+				}
+			}
+		}
+		response.pipe(stream);
+		await events.once(stream, "finish");
+
+		let filename = await libFileOps.findUnusedName(".", args.save, ".zip");
+		await fs.rename(tempFilename, filename);
+
+		logger.info(`Downloaded ${args.save}${args.save === filename ? "" : ` as ${filename}`}`);
+	},
+}));
+
 instanceCommands.add(new libCommand.Command({
 	definition: ["delete-save <instance> <save>", "Delete a save from an instance", (yargs) => {
 		yargs.positional("instance", { describe: "Instance to delete save from", type: "string" });

packages/lib/link/messages.js

@@ -543,6 +543,19 @@ messages.deleteSave = new Request({
 	},
 });
 
+messages.downloadSave = new Request({
+	type: "download_save",
+	links: ["control-master"],
+	permission: "core.instance.save.download",
+	requestProperties: {
+		"instance_id": { type: "integer" },
+		"save": { type: "string" },
+	},
+	responseProperties: {
+		"stream_id": { type: "string" },
+	},
+});
+
 messages.pullSave = new Request({
 	type: "pull_save",
 	links: ["master-slave"],
@@ -556,6 +569,16 @@ messages.pullSave = new Request({
 	},
 });
 
+messages.pushSave = new Request({
+	type: "push_save",
+	links: ["master-slave"],
+	requestProperties: {
+		"instance_id": { type: "integer" },
+		"stream_id": { type: "string" },
+		"save": { type: "string" },
+	},
+});
+
 messages.setSaveListSubscriptions = new Request({
 	type: "set_save_list_subscriptions",
 	links: ["control-master"],

packages/lib/users.js

@@ -391,6 +391,11 @@ definePermission({
 	title: "Upload save to instance",
 	description: "Upload savegames to instances.",
 });
+definePermission({
+	name: "core.instance.save.download",
+	title: "Download save from instance",
+	description: "Download savegames from instances.",
+});
 definePermission({
 	name: "core.instance.save.list_subscribe",
 	title: "Subscribe to save list updatess",

packages/master/src/ControlConnection.js

@@ -13,6 +13,7 @@ const libPrometheus = require("@clusterio/lib/prometheus");
 const libUsers = require("@clusterio/lib/users");
 
 const BaseConnection = require("./BaseConnection");
+const routes = require("./routes");
 
 const lastQueryLogTime = new libPrometheus.Gauge(
 	"clusterio_master_last_query_log_duration_seconds",
@@ -344,6 +345,29 @@ class ControlConnection extends BaseConnection {
 		}
 	}
 
+	async downloadSaveRequestHandler(message) {
+		let { instance_id, save } = message.data;
+		let stream = await routes.createProxyStream(this._master.app);
+		stream.filename = save;
+
+		let ready = new Promise((resolve, reject) => {
+			stream.events.on("source", resolve);
+			stream.events.on("timeout", () => reject(
+				new libErrors.RequestError("Timed out establishing stream from slave")
+			));
+		});
+		ready.catch(() => {});
+
+		let result = await this._master.forwardRequestToInstance(libLink.messages.pushSave, {
+			instance_id,
+			stream_id: stream.id,
+			save,
+		});
+
+		await ready;
+		return { stream_id: stream.id };
+	}
+
 	async setLogSubscriptionsRequestHandler(message) {
 		this.logSubscriptions = message.data;
 		this.updateLogSubscriptions();

packages/master/src/routes.js

@@ -185,6 +185,7 @@ async function createProxyStream(app) {
 		flowing: false,
 		size: null,
 		mime: null,
+		filename: null,
 		events: new events.EventEmitter(),
 		timeout: setTimeout(() => {
 			stream.events.emit("timeout");
@@ -201,6 +202,27 @@ async function createProxyStream(app) {
 	return stream;
 }
 
+async function putStream(req, res) {
+	let stream = req.app.locals.streams.get(req.params.id);
+	if (!stream || stream.source) {
+		res.sendStatus(404);
+		return;
+	}
+	stream.source = req;
+	stream.mime = req.get("Content-Type");
+	stream.size = req.get("Content-Length");
+
+	stream.events.emit("source");
+	stream.events.on("close", () => {
+		if (!stream.flowing) {
+			req.resume();
+			res.sendStatus(500);
+		} else {
+			res.sendStatus(200);
+		}
+	});
+}
+
 async function getStream(req, res) {
 	let stream = req.app.locals.streams.get(req.params.id);
 	if (!stream || stream.flowing) {
@@ -213,6 +235,11 @@ async function getStream(req, res) {
 		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("finish", () => {
@@ -369,6 +396,7 @@ function addRouteHandlers(app) {
 		validateSlaveToken,
 		(req, res, next) => 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.post("/api/upload-save",
 		validateUserToken,

packages/slave/slave.js

@@ -1461,6 +1461,38 @@ class Slave extends libLink.Link {
 		return { save: filename };
 	}
 
+	async pushSaveRequestHandler(message) {
+		let { instance_id, stream_id, save } = message.data;
+		try {
+			checkFilename(save);
+		} catch (err) {
+			throw new libErrors.RequestError(`Save name ${err.message}`);
+		}
+		let instanceInfo = this.instanceInfos.get(instance_id);
+		if (!instanceInfo) {
+			throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`);
+		}
+
+		let content;
+		try {
+			// phin doesn't support streaming requests :(
+			content = await fs.readFile(path.join(instanceInfo.path, "saves", save));
+		} catch (err) {
+			if (err.code === "ENOENT") {
+				throw new libErrors.RequestError(`${save} does not exist`);
+			}
+			throw err;
+		}
+
+		let url = new URL(this.config.get("slave.master_url"));
+		url.pathname += `api/stream/${stream_id}`;
+		phin({
+			url, method: "PUT",
+			core: { ca: this.tlsCa },
+			data: content,
+		}).catch(err => logger.error(`Error pushing save to master:\n${err.stack}`));
+	}
+
 	async exportDataRequestHandler(message, request) {
 		let instanceId = message.data.instance_id;
 		let instanceConnection = await this._connectInstance(instanceId);

packages/web_ui/src/components/SavesList.jsx

@@ -77,6 +77,19 @@ export default function SavesList(props) {
 						);
 					}}
 				>Load save</Button>
+				<Button
+					onClick={() => {
+						libLink.messages.downloadSave.send(
+							control, { instance_id: props.instance.id, save: save.name }
+						).then(response => {
+							let url = new URL(webRoot, document.location);
+							url.pathname += `api/stream/${response.stream_id}`;
+							document.location = url;
+						}).catch(
+							notifyErrorHandler("Error downloading save")
+						);
+					}}
+				>Download</Button>
 				<Popconfirm
 					title="Permanently delete save?"
 					okText="Delete"

test/integration/clusterio.js

@@ -377,6 +377,20 @@ describe("Integration of Clusterio", function() {
 			});
 		});
 
+		describe("instance download-save", function() {
+			it("should download a save", async function() {
+				await fs.remove(path.join("temp", "test", "upload.zip"));
+				await execCtl("instance download-save 44 upload.zip");
+				assert(await fs.pathExists(path.join("temp", "test", "upload.zip")));
+			});
+			it("should error if save does not exist", async function() {
+				await assert.rejects(execCtl("instance download-save 44 invalid"));
+			});
+			it("should error on path traversal attacks", async function() {
+				await assert.rejects(execCtl("instance download-save 44 ../factorio-current.log"));
+			});
+		});
+
 		describe("instance delete-save", function() {
 			it("should delete a save", async function() {
 				await execCtl("instance delete-save 44 upload.zip");

test/master/routes.js

@@ -6,6 +6,7 @@ const jwt = require("jsonwebtoken");
 const http = require("http");
 const phin = require("phin");
 
+const { wait } = require("@clusterio/lib/helpers");
 const routes = require("@clusterio/master/src/routes");
 const mock = require("../mock");
 
@@ -27,7 +28,13 @@ describe("master/src/routes", function() {
 	});
 	describe("/api/stream/:id", function() {
 		it("should respond with 404 if stream doesn't exist", async function() {
-			let response = await phin(`http://localhost:${port}/api/stream/01020304`);
+			let response;
+			response = await phin(`http://localhost:${port}/api/stream/01020304`);
+			assert.equal(response.statusCode, 404);
+			response = await phin({
+				url: `http://localhost:${port}/api/stream/01020304`, method: "PUT",
+				data: "test stream",
+			});
 			assert.equal(response.statusCode, 404);
 		});
 		it("should respond with 500 if stream times out", async function() {
@@ -35,6 +42,38 @@ describe("master/src/routes", function() {
 			let response = await phin(`http://localhost:${port}/api/stream/${stream.id}`);
 			assert.equal(response.statusCode, 500);
 		});
+		it("should passthrough a stream", async function() {
+			let stream;
+			let responses;
+			stream = await routes.createProxyStream(master.app);
+			responses = await Promise.all([
+				(async function() {
+					await wait(100);
+					return await phin({
+						url: `http://localhost:${port}/api/stream/${stream.id}`, method: "PUT",
+						data: "test content",
+					});
+				}()),
+				phin(`http://localhost:${port}/api/stream/${stream.id}`),
+			]);
+			assert.equal(responses[0].statusCode, 200);
+			assert.equal(responses[1].statusCode, 200);
+			assert.equal(responses[1].body.toString(), "test content");
+			stream = await routes.createProxyStream(master.app);
+			responses = await Promise.all([
+				phin({
+					url: `http://localhost:${port}/api/stream/${stream.id}`, method: "PUT",
+					data: "test content",
+				}),
+				(async function() {
+					await wait(100);
+					return await phin(`http://localhost:${port}/api/stream/${stream.id}`);
+				}()),
+			]);
+			assert.equal(responses[0].statusCode, 200);
+			assert.equal(responses[1].statusCode, 200);
+			assert.equal(responses[1].body.toString(), "test content");
+		});
 	});
 	describe("/api/upload-save", function() {
 		it("should respond with 401 with invalid token", async function() {