Stop reading all rejected uploads and close stalled transfers

Guard the export, save, mod and stream routes with one middleware that
closes the connection of an upload answered before its body was read,
and of a transfer that has made no progress for the new
controller.transfer_idle_timeout.

The save and mod uploads share one handler that responds on the first
error instead of reading the rest of the upload, and uploads cut short
now fail the transfer instead of leaving it hanging.

Node's 5 minute limit on receiving a request is turned off so large
uploads over slow connections are not cut off.

docs/configuration.md

@@ -231,6 +231,17 @@ Must be greater than 0.
 Defaults to 15.
 
 
+### controller.transfer_idle_timeout
+
+Time in seconds an upload to or download from the controller can go without any data being transferred before the controller closes it.
+This applies to saves, mods and exports uploaded to the controller and to proxy streams through it.
+There is no limit on how long a transfer can take as long as it keeps making progress.
+Time spent waiting for a proxy stream to start also counts, so this should be higher than controller.proxy_stream_timeout.
+Must be greater than 0.
+
+Defaults to 60.
+
+
 ### controller.factorio_username
 
 Username of the Factorio.com account to authenticate with the Factorio API with.

packages/controller/src/Controller.ts

@@ -473,7 +473,7 @@ export default class Controller {
 
 		// Only start listening for connections after all plugins have loaded
 		if (httpPort) {
-			this.httpServer = http.createServer(this.app);
+			this.httpServer = http.createServer(routes.httpServerOptions, 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() as AddressInfo).port}`);
@@ -492,6 +492,7 @@ export default class Controller {
 			}
 
 			this.httpsServer = https.createServer({
+				...routes.httpServerOptions,
 				key: privateKey,
 				cert: certificate,
 			}, this.app);

packages/controller/src/routes.ts

@@ -445,41 +445,62 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
 	return stream;
 }
 
-/** Time a rejected upload is read and thrown away before its connection is closed */
-const rejectedUploadDrainMs = 1000;
+/**
+ * Options for the HTTP servers the routes are served from
+ *
+ * Node closes requests that have not been received in full after 5 minutes
+ * by default, which would cut off large uploads over slow connections.
+ * Transfers are instead closed when they stop making progress.
+ */
+export const httpServerOptions = { requestTimeout: 0 };
+
+/** Time an upload is read and thrown away after it was answered, before its connection is closed */
+const answeredUploadDrainMs = 1000;
 
 /**
- * Reject an upload without reading the rest of its body
+ * Guard a route transferring files against wasted and stalled transfers
  *
  * Node reads and discards the body of a request answered before it was
  * read to keep the connection open. A client, or a proxy in front of the
  * controller, that keeps sending would upload the whole body for nothing,
  * so the connection is closed if the body hasn't ended shortly after the
  * response. Closing it right away could make the client lose the response.
- * @param req - The upload.
- * @param res - Response to the upload.
- * @param status - HTTP status to respond with.
+ *
+ * The connection is also closed if no data was sent or received on it for
+ * controller.transfer_idle_timeout, so that a transfer which stopped
+ * making progress does not linger while a slow one can take its time.
+ * @param req - Request to guard.
+ * @param res - Response to the request.
+ * @param next - Next handler for the route.
  */
-function rejectUpload(req: Request, res: Response, status: number) {
-	res.sendStatus(status);
-	if (req.complete) {
-		return;
-	}
+function guardTransfer(req: Request, res: Response, next: any) {
 	const socket = req.socket;
-	const timeout = setTimeout(() => socket.destroy(), rejectedUploadDrainMs);
-	const onClose = () => clearTimeout(timeout);
-	socket.once("close", onClose);
-	// The connection may be kept open for more requests, so don't leave a listener on it
-	req.once("end", () => {
-		clearTimeout(timeout);
-		socket.off("close", onClose);
+	const idleTimeout = req.app.locals.controller.config.get("controller.transfer_idle_timeout");
+	// Node sets the timeout for the next request on the connection itself.
+	res.setTimeout(idleTimeout * 1000, () => {
+		logger.warn(`Closing ${req.method} ${req.path} after ${idleTimeout} seconds without progress`);
+		socket.destroy();
 	});
+	res.once("finish", () => {
+		if (req.complete || socket.destroyed) {
+			return;
+		}
+		const timeout = setTimeout(() => socket.destroy(), answeredUploadDrainMs);
+		const onClose = () => clearTimeout(timeout);
+		socket.once("close", onClose);
+		// The connection may be kept open for more requests, so don't leave a listener on it
+		req.once("end", () => {
+			clearTimeout(timeout);
+			socket.off("close", onClose);
+		});
+	});
+	next();
 }
 
 async function putStream(req: Request<{ id: string }>, res: Response) {
 	const stream = req.app.locals.streams.get(req.params.id);
 	if (!stream || stream.source) {
-		rejectUpload(req, res, 404);
+		res.sendStatus(404);
 		return;
 	}
 	stream.source = req;
@@ -488,11 +509,8 @@ async function putStream(req: Request<{ id: string }>, res: Response) {
 
 	stream.events.emit("source");
 	stream.events.on("close", () => {
-		if (!stream.flowing) {
-			rejectUpload(req, res, 500);
-		} else {
-			res.sendStatus(200);
-		}
+		// The stream also closes when it timed out or the receiver disconnected.
+		res.sendStatus(stream.flowing && req.complete ? 200 : 500);
 	});
 }
 
@@ -506,9 +524,19 @@ function startStream(res: Response, stream: ProxyStream) {
 	} else {
 		res.append("Content-Disposition", "attachment");
 	}
-	stream.source!.pipe(res);
+	const source = stream.source!;
+	source.pipe(res);
 	stream.flowing = true;
+	// Pipe does not pass on a source that was cut short, the receiver would wait forever.
+	const onSourceClose = () => {
+		if (!res.writableEnded) {
+			res.destroy();
+		}
+	};
+	source.once("close", onSourceClose);
 	res.on("close", () => {
+		source.off("close", onSourceClose);
+		source.unpipe(res);
 		stream.events.emit("close");
 	});
 	clearTimeout(stream.timeout);
@@ -539,6 +567,141 @@ const zipMimes = [
 
 const contentTypeRegExp = /^([!#$%&'*+\-.^_`|~0-9A-Za-z]+\/[!#$%&'*+\-.^_`|~0-9A-Za-z]+)/;
 
+/** Error in an upload that is the fault of the request */
+class UploadRequestError extends Error {
+	constructor(message: string, public status = 400) {
+		super(message);
+	}
+}
+
+interface UploadedZip {
+	/** Content of the file, ends with an error if the upload is cut short */
+	stream: nodeStream.Readable;
+	filename: string;
+	mime: string;
+	/** Form fields received before the file, or the query parameters of a zip upload */
+	fields: Map<string, string>;
+	/** True if the file is part of a multipart form */
+	multipart: boolean;
+}
+
+/**
+ * Receive zip files uploaded as either the request body or a multipart form
+ *
+ * Responds with the first error, without reading the rest of the upload.
+ * Errors other than UploadRequestError are logged and responded to with 500.
+ * @param req - The upload.
+ * @param res - Response to the upload.
+ * @param kind - What is uploaded, for the log.
+ * @param handleFile -
+ *     Called with each file, the next file is not read before the stream of
+ *     the previous has been consumed.
+ * @returns true if all files were handled, false if an error was responded with.
+ */
+async function receiveZipUpload(
+	req: Request,
+	res: Response,
+	kind: string,
+	handleFile: (file: UploadedZip) => Promise<void>,
+): Promise<boolean> {
+	function fail(err: any) {
+		if (res.headersSent) {
+			return;
+		}
+		if (err instanceof UploadRequestError) {
+			res.status(err.status).json({ request_errors: [err.message] });
+			return;
+		}
+		logger.error(`Error uploading ${kind}: ${err.message}`);
+		res.status(500).json({ errors: [err.message], request_errors: [] });
+	}
+
+	const match = contentTypeRegExp.exec(req.get("Content-Type") || "");
+	const contentMime = match ? match[1].toLowerCase() : "";
+
+	if (zipMimes.includes(contentMime)) {
+		try {
+			const filename = req.query.filename;
+			if (typeof filename !== "string") {
+				throw new UploadRequestError("Missing or invalid filename parameter");
+			}
+			if (!filename.endsWith(".zip")) {
+				throw new UploadRequestError("filename must end with .zip");
+			}
+			const fields = new Map<string, string>();
+			for (const [name, value] of Object.entries(req.query)) {
+				if (typeof value === "string") {
+					fields.set(name, value);
+				}
+			}
+			await handleFile({ stream: req, filename, mime: contentMime, fields, multipart: false });
+		} catch (err: any) {
+			fail(err);
+			return false;
+		}
+		return true;
+	}
+
+	if (contentMime !== "multipart/form-data") {
+		fail(new UploadRequestError("invalid Content-Type", 415));
+		return false;
+	}
+
+	let parser: busboy.Busboy;
+	try {
+		parser = busboy({ headers: req.headers });
+	} catch (err: any) {
+		fail(new UploadRequestError(err.message));
+		return false;
+	}
+
+	return await new Promise<boolean>(resolve => {
+		const fields = new Map<string, string>();
+		const tasks: Promise<void>[] = [];
+		let failed = false;
+		function onError(err: any) {
+			if (failed) {
+				return;
+			}
+			failed = true;
+			fail(err);
+			// Stop reading the upload, this ends the file being read with an error.
+			req.unpipe(parser);
+			parser.destroy();
+			resolve(false);
+		}
+		parser.on("field", (name, value) => {
+			fields.set(name, value);
+		});
+		parser.on("file", (name, stream, { filename, mimeType }) => {
+			// The handler may be done with the stream by the time it fails.
+			stream.on("error", () => {});
+			if (failed) {
+				stream.resume();
+			} else if (!zipMimes.includes(mimeType)) {
+				onError(new UploadRequestError("invalid file Content-Type"));
+			} else if (!filename?.endsWith(".zip")) {
+				onError(new UploadRequestError("filename must end with .zip"));
+			} else {
+				tasks.push(handleFile({
+					stream, filename, mime: mimeType, fields: new Map(fields), multipart: true,
+				}).catch(onError));
+			}
+		});
+		parser.on("error", onError);
+		parser.on("close", () => {
+			Promise.all(tasks).then(() => resolve(!failed));
+		});
+		// Pipe does not end the parser if the upload is cut short.
+		req.once("close", () => {
+			if (!req.complete) {
+				onError(new Error("Upload was aborted"));
+			}
+		});
+		req.pipe(parser);
+	});
+}
+
 async function uploadSave(req: Request, res: Response) {
 	try {
 		res.locals.user.checkPermission("core.instance.save.upload");
@@ -547,26 +710,24 @@ async function uploadSave(req: Request, res: Response) {
 		return;
 	}
 
-	let contentType = req.get("Content-Type") || "";
-	let match = contentTypeRegExp.exec(contentType);
-	if (!match) {
-		res.status(415).json({ request_errors: ["invalid Content-Type"] });
-		return;
-	}
-	let contentMime = match[1].toLowerCase();
-
-	let tasks = [];
-	let errors: string[] = [];
-	let requestErrors = [];
 	let saves: string[] = [];
+	async function handleFile(file: UploadedZip) {
+		const value = file.fields.get("instance_id");
+		if (value === undefined && file.multipart) {
+			throw new UploadRequestError("instance_id must come before files uploaded");
+		}
+		const instanceId = Number.parseInt(String(value), 10);
+		if (Number.isNaN(instanceId)) {
+			throw new UploadRequestError(
+				file.multipart ? "invalid instance_id" : "Missing or invalid instance_id parameter"
+			);
+		}
 
-	async function handleFile(instanceId:number, stream: nodeStream.Readable, filename: string, streamMime: string) {
 		let proxyStream = await createProxyStream(req.app);
-		proxyStream.source = stream;
-		proxyStream.mime = streamMime;
+		proxyStream.source = file.stream;
+		proxyStream.mime = file.mime;
 		let timeout = new Promise<never>((_, reject) => {
 			proxyStream.events.on("timeout", () => {
-				stream.resume();
 				reject(new Error("Timed out establishing stream to host"));
 			});
 		});
@@ -577,7 +738,7 @@ async function uploadSave(req: Request, res: Response) {
 					new lib.InstancePullSaveRequest(
 						instanceId,
 						proxyStream.id,
-						filename,
+						file.filename,
 					)
 				),
 				timeout,
@@ -586,90 +747,13 @@ async function uploadSave(req: Request, res: Response) {
 
 		} catch (err: any) {
 			proxyStream.events.emit("close");
-			logger.error(`Error uploading save: ${err.message}`);
-			errors.push(err.message);
-			stream.resume();
+			throw err;
 		}
 	}
 
-	if (contentMime === "multipart/form-data") {
-		await new Promise(resolve => {
-			let instanceId: number | undefined;
-			function fileHandler(
-				name: string,
-				stream: nodeStream.Readable,
-				{ filename, mimeType }: { filename: string, mimeType: string }
-			) {
-				if (instanceId === undefined) {
-					requestErrors.push("instance_id must come before files uploaded");
-				}
-				if (!zipMimes.includes(mimeType)) {
-					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(instanceId as number, stream, filename, mimeType));
-			}
-			let parser = busboy({ headers: req.headers });
-			parser.on("file", fileHandler);
-			parser.on("field", (name, value, info) => {
-				if (name === "instance_id") {
-					instanceId = Number.parseInt(value, 10);
-					if (Number.isNaN(instanceId)) {
-						requestErrors.push("invalid instance_id");
-					}
-				}
-			});
-			parser.on("close", resolve);
-			parser.on("error", (err: any) => {
-				logger.error(`Error parsing multipart request in upload-save:\n${err.stack}`);
-				errors.push(err.message);
-			});
-			req.pipe(parser);
-		});
-
-	} else if (zipMimes.includes(contentMime)) {
-		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(String(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 as string, contentMime));
-		}
-
-	} else {
-		res.status(415).json({ request_errors: ["invalid Content-Type"] });
-		return;
+	if (await receiveZipUpload(req, res, "save", handleFile)) {
+		res.json({ saves });
 	}
-
-	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 checkModName(name: string) {
@@ -688,37 +772,18 @@ async function uploadMod(req: Request, res: Response) {
 		return;
 	}
 
-	let contentType = req.get("Content-Type")!;
-	let match = contentTypeRegExp.exec(contentType);
-	if (!match) {
-		res.status(415).json({ request_errors: ["invalid Content-Type"] });
-		return;
-	}
-	let contentMime = match[1].toLowerCase();
-
-	let tasks = [];
-	let errors: string[] = [];
-	let requestErrors = [];
 	let mods: any[] = [];
-
-	async function handleFile(stream: nodeStream.Readable, filename: string) {
-		try {
-			checkModName(filename);
-
-		} catch (err: any) {
-			logger.error(`Error uploading mod: ${err.message}`);
-			errors.push(err.message);
-			stream.resume();
-			return;
-		}
+	async function handleFile(file: UploadedZip) {
+		checkModName(file.filename);
 
 		const modsDirectory = req.app.locals.controller.config.get("controller.mods_directory");
-		const filePath = path.join(modsDirectory, filename);
+		const filePath = path.join(modsDirectory, file.filename);
 		const [tempFilePath, writeStream] = await lib.createTempWriteStream(filePath);
 
 		try {
-			stream.pipe(writeStream);
-			await finished(writeStream);
+			file.stream.pipe(writeStream);
+			// Pipe does not pass on an upload that was cut short.
+			await Promise.all([finished(file.stream), finished(writeStream)]);
 
 			const modInfo = await lib.ModInfo.fromModFile(tempFilePath);
 			checkModName(modInfo.filename);
@@ -727,11 +792,8 @@ async function uploadMod(req: Request, res: Response) {
 			mods.push(modInfo.toJSON());
 
 		} catch (err: any) {
-			logger.error(`Error uploading mod: ${err.message}`);
-			errors.push(err.message);
-			stream.resume();
-
 			// Attempt to clean up.
+			file.stream.unpipe(writeStream);
 			writeStream.destroy();
 
 			try {
@@ -741,72 +803,13 @@ async function uploadMod(req: Request, res: Response) {
 					logger.error(`Error removing ${tempFilePath}: ${unlinkErr.message}`);
 				}
 			}
+			throw err;
 		}
 	}
 
-	if (contentMime === "multipart/form-data") {
-		await new Promise(resolve => {
-			function fileHandler(
-				name: string,
-				stream: nodeStream.Readable,
-				{ filename, mimeType }: { filename: string, mimeType: string }
-			) {
-				if (!zipMimes.includes(mimeType)) {
-					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(stream, filename));
-			}
-
-			let parser = busboy({ headers: req.headers });
-			parser.on("file", fileHandler);
-			parser.on("close", resolve);
-			parser.on("error", (err: any) => {
-				logger.error(`Error parsing multipart request in upload-mod:\n${err.stack}`);
-				errors.push(err.message);
-			});
-			req.pipe(parser);
-		});
-
-	} else if (zipMimes.includes(contentMime)) {
-		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");
-		}
-
-		if (errors.length || requestErrors.length) {
-			req.resume();
-		} else {
-			tasks.push(handleFile(req, filename as string));
-		}
-
-	} else {
-		res.status(415).json({ request_errors: ["invalid Content-Type"] });
-		return;
+	if (await receiveZipUpload(req, res, "mod", handleFile)) {
+		res.json({ mods });
 	}
-
-	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({ mods });
 }
 
 
@@ -830,20 +833,25 @@ export function addRouteHandlers(app: Application) {
 		res.json({ valid: true });
 	});
 	app.put("/api/upload-export",
+		guardTransfer,
 		validateHostToken,
 		(req:Request, res:Response, next:any) => uploadExport(req, res).catch(next)
 	);
 	app.put("/api/stream/:id",
+		guardTransfer,
 		(req:Request<{ id: string }>, res:Response, next:any) => putStream(req, res).catch(next)
 	);
 	app.get("/api/stream/:id",
+		guardTransfer,
 		(req:Request<{ id: string }>, res:Response, next:any) => getStream(req, res).catch(next)
 	);
 	app.post("/api/upload-save",
+		guardTransfer,
 		validateUserToken,
 		(req:Request, res:Response, next:any) => uploadSave(req, res).catch(next)
 	);
 	app.post("/api/upload-mod",
+		guardTransfer,
 		validateUserToken,
 		(req:Request, res:Response, next:any) => uploadMod(req, res).catch(next)
 	);

packages/lib/src/config/definitions.ts

@@ -31,6 +31,7 @@ export interface ControllerConfigFields {
 	"controller.max_subscription_filters": number;
 	"controller.system_metrics_interval": number;
 	"controller.proxy_stream_timeout": number;
+	"controller.transfer_idle_timeout": number;
 	"controller.factorio_username": string | null;
 	"controller.factorio_token": string | null;
 	"controller.share_factorio_credentials_with_hosts": boolean;
@@ -215,6 +216,13 @@ export class ControllerConfig extends classes.Config<ControllerConfigFields> {
 			initialValue: 15,
 			validator: validators.greaterThanZero,
 		},
+		"controller.transfer_idle_timeout": {
+			title: "Transfer Idle Timeout",
+			description: "Time in seconds an upload or download can go without progress before it's closed.",
+			type: "number",
+			initialValue: 60,
+			validator: validators.greaterThanZero,
+		},
 		"controller.session_timeout": {
 			title: "Session Timeout",
 			description: "Time in seconds before giving up resuming a dropped WebSocket session.",

test/controller/routes.js

@@ -12,6 +12,7 @@ import { HostPairingManager, HostRecord } from "@clusterio/controller";
 
 import * as routes from "@clusterio/controller/dist/node/src/routes.js";
 import * as mock from "../mock.js";
+import { createZipBuffer } from "../common.js";
 
 import "../setup_logging.js";
 
@@ -22,7 +23,7 @@ describe("controller/src/routes", function() {
 	beforeEach(async function() {
 		controller = new mock.MockController();
 		routes.addRouteHandlers(controller.app);
-		server = http.createServer(controller.app);
+		server = http.createServer(routes.httpServerOptions, controller.app);
 		server.listen(0, "localhost");
 		await events.once(server, "listening");
 		port = server.address().port;
@@ -30,6 +31,67 @@ describe("controller/src/routes", function() {
 	afterEach(function() {
 		server.close();
 	});
+
+	const largeSize = 64 * 1024 * 1024;
+	/**
+	 * Send a large body at about 6 MiB/s whether or not the server reads it,
+	 * the way a proxy in front of the controller passes on an upload, until
+	 * the connection closes, and count what the server read.
+	 * @param {string} url - URL to send to.
+	 * @param {object} options - Options for the request.
+	 * @param {string=} options.method - HTTP method, defaults to PUT.
+	 * @param {object=} options.headers - Headers to send.
+	 * @param {string=} options.prefix - Content to send before the large body.
+	 * @returns {Promise<{ status: number|undefined, body: string, serverBytesRead: number }>} result
+	 */
+	async function sendLargeBody(url, { method = "PUT", headers = {}, prefix = "" } = {}) {
+		const sockets = [];
+		const onConnection = socket => sockets.push(socket);
+		server.on("connection", onConnection);
+		const request = http.request(url, {
+			method, headers: { "Content-Length": Buffer.byteLength(prefix) + largeSize, ...headers },
+		});
+		let status;
+		let body = "";
+		request.on("response", response => {
+			status = response.statusCode;
+			response.setEncoding("utf8");
+			response.on("data", chunk => { body += chunk; });
+			response.on("error", () => {});
+		});
+		// The server may close the connection while the body is being sent
+		request.on("error", () => {});
+		const closed = new Promise(resolve => request.on("close", resolve));
+		request.write(prefix);
+		const chunk = Buffer.alloc(64 * 1024);
+		for (let sent = 0; sent < largeSize && !request.closed; sent += chunk.length) {
+			request.write(chunk);
+			await lib.wait(10);
+		}
+		if (!request.closed) {
+			request.end();
+		}
+		await closed;
+		server.off("connection", onConnection);
+		await Promise.all(sockets.map(
+			socket => socket.destroyed || new Promise(resolve => socket.on("close", resolve))
+		));
+		return { status, body, serverBytesRead: sockets.reduce((total, socket) => total + socket.bytesRead, 0) };
+	}
+	function assertNotRead(result) {
+		assert(
+			result.serverBytesRead < largeSize / 4,
+			`server read ${result.serverBytesRead} of ${largeSize} bytes`
+		);
+	}
+	function multipartPrefix(boundary, fields, filename, mime = "application/zip") {
+		let prefix = "";
+		for (const [name, value] of Object.entries(fields)) {
+			prefix += `--${boundary}\r\nContent-Disposition: form-data; name="${name}"\r\n\r\n${value}\r\n`;
+		}
+		return `${prefix}--${boundary}\r\nContent-Disposition: form-data; name="file"; filename="${filename}"\r\n` +
+			`Content-Type: ${mime}\r\n\r\n`;
+	}
 	describe("/api/stream/:id", function() {
 		let endpoint;
 		beforeEach(function() {
@@ -49,57 +111,78 @@ describe("controller/src/routes", function() {
 			let response = await fetch(`${endpoint}/${stream.id}`);
 			assert.equal(response.status, 500);
 		});
-		/**
-		 * PUT a body at about 6 MiB/s whether or not the server reads it, the way
-		 * a proxy in front of the controller passes on an upload, until the
-		 * connection closes, and count what the server read.
-		 * @param {string} url - URL to PUT to.
-		 * @param {number} size - Size of the body in bytes.
-		 * @returns {Promise<{ status: number|undefined, serverBytesRead: number }>} result
-		 */
-		async function putLargeBody(url, size) {
-			const sockets = [];
-			const onConnection = socket => sockets.push(socket);
-			server.on("connection", onConnection);
-			const request = http.request(url, { method: "PUT", headers: { "Content-Length": size } });
-			let status;
-			request.on("response", response => {
-				status = response.statusCode;
-				response.on("error", () => {});
-				response.resume();
-			});
-			// The server may close the connection while the body is being sent
-			request.on("error", () => {});
-			const closed = new Promise(resolve => request.on("close", resolve));
-			const chunk = Buffer.alloc(64 * 1024);
-			for (let sent = 0; sent < size && !request.closed; sent += chunk.length) {
-				request.write(chunk);
-				await wait(10);
-			}
-			if (!request.closed) {
-				request.end();
-			}
-			await closed;
-			server.off("connection", onConnection);
-			await Promise.all(sockets.map(
-				socket => socket.destroyed || new Promise(resolve => socket.on("close", resolve))
-			));
-			return { status, serverBytesRead: sockets.reduce((total, socket) => total + socket.bytesRead, 0) };
-		}
 		it("should not read the body of a PUT to a stream that doesn't exist", async function() {
 			this.timeout(30000);
-			const size = 64 * 1024 * 1024;
-			const result = await putLargeBody(`${endpoint}/01020304`, size);
+			const result = await sendLargeBody(`${endpoint}/01020304`);
 			assert.equal(result.status, 404);
-			assert(result.serverBytesRead < size / 4, `server read ${result.serverBytesRead} of ${size} bytes`);
+			assertNotRead(result);
 		});
 		it("should not read the body of a PUT to a stream that times out", async function() {
 			this.timeout(30000);
-			const size = 64 * 1024 * 1024;
 			const stream = await routes.createProxyStream(controller.app);
-			const result = await putLargeBody(`${endpoint}/${stream.id}`, size);
+			const result = await sendLargeBody(`${endpoint}/${stream.id}`);
 			assert.equal(result.status, 500);
-			assert(result.serverBytesRead < size / 4, `server read ${result.serverBytesRead} of ${size} bytes`);
+			assertNotRead(result);
+		});
+		it("should respond with 500 and stop reading when the receiver disconnects", async function() {
+			this.timeout(30000);
+			const stream = await routes.createProxyStream(controller.app);
+			const upload = sendLargeBody(`${endpoint}/${stream.id}`);
+			const abort = new AbortController();
+			const response = await fetch(`${endpoint}/${stream.id}`, { signal: abort.signal });
+			assert.equal(response.status, 200);
+			await response.body.getReader().read();
+			abort.abort();
+			const result = await upload;
+			assert.equal(result.status, 500);
+			assertNotRead(result);
+		});
+		it("should close the receiver when the upload is cut short", async function() {
+			const stream = await routes.createProxyStream(controller.app);
+			const request = http.request(
+				`${endpoint}/${stream.id}`, { method: "PUT", headers: { "Content-Length": 1024 * 1024 } }
+			);
+			request.on("error", () => {});
+			request.write(Buffer.alloc(1024));
+			const response = await fetch(`${endpoint}/${stream.id}`);
+			assert.equal(response.status, 200);
+			request.destroy();
+			await assert.rejects(response.arrayBuffer());
+		});
+		it("should close a transfer that stops making progress", async function() {
+			controller.mockConfigEntries.set("controller.transfer_idle_timeout", 0.5);
+			const stream = await routes.createProxyStream(controller.app);
+			const request = http.request(
+				`${endpoint}/${stream.id}`, { method: "PUT", headers: { "Content-Length": 1024 * 1024 } }
+			);
+			request.on("error", () => {});
+			const closed = new Promise(resolve => request.on("close", resolve));
+			request.write(Buffer.alloc(1024));
+			const response = await fetch(`${endpoint}/${stream.id}`);
+			assert.equal(response.status, 200);
+			const start = Date.now();
+			await assert.rejects(response.arrayBuffer());
+			await closed;
+			assert(Date.now() - start < 1500, `closed after ${Date.now() - start} ms`);
+		});
+		it("should not close a slow transfer that makes progress", async function() {
+			this.timeout(10000);
+			controller.mockConfigEntries.set("controller.transfer_idle_timeout", 0.5);
+			const stream = await routes.createProxyStream(controller.app);
+			const request = http.request(
+				`${endpoint}/${stream.id}`, { method: "PUT", headers: { "Content-Length": 10 * 1024 } }
+			);
+			const uploaded = events.once(request, "response");
+			const download = fetch(`${endpoint}/${stream.id}`).then(response => response.arrayBuffer());
+			for (let i = 0; i < 10; i++) {
+				request.write(Buffer.alloc(1024));
+				await lib.wait(200);
+			}
+			request.end();
+			const [response] = await uploaded;
+			response.resume();
+			assert.equal(response.statusCode, 200);
+			assert.equal((await download).byteLength, 10 * 1024);
 		});
 		it("should passthrough a stream", async function() {
 			let stream;
@@ -860,5 +943,253 @@ describe("controller/src/routes", function() {
 			assert.deepEqual(await response.json(), { saves: ["test.zip"] });
 			assert.equal(response.status, 200);
 		});
+
+		function uploadVariants() {
+			return [
+				{
+					url: `${endpoint}?instance_id=123&filename=test.zip`,
+					headers: { "Content-Type": "application/zip" },
+				},
+				{
+					url: endpoint,
+					headers: { "Content-Type": "multipart/form-data; boundary=test" },
+					prefix: multipartPrefix("test", { instance_id: "123" }, "test.zip"),
+				},
+			];
+		}
+		it("should not read the body of an upload without a valid token", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(`${endpoint}?instance_id=123&filename=file.zip`, {
+				method: "POST", headers: { "Content-Type": "application/zip" },
+			});
+			assert.equal(result.status, 401);
+			assertNotRead(result);
+		});
+		it("should not read the body of an invalid file request", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(`${endpoint}?instance_id=123&filename=test.txt`, {
+				method: "POST", headers: { "Content-Type": "application/zip", "X-Access-Token": testToken },
+			});
+			assert.equal(result.status, 400);
+			assert.deepEqual(JSON.parse(result.body), { request_errors: ["filename must end with .zip"] });
+			assertNotRead(result);
+		});
+		it("should not read the body of an invalid multipart request", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(endpoint, {
+				method: "POST",
+				headers: { "Content-Type": "multipart/form-data; boundary=test", "X-Access-Token": testToken },
+				prefix: multipartPrefix("test", { instance_id: "invalid" }, "test.zip"),
+			});
+			assert.equal(result.status, 400);
+			assert.deepEqual(JSON.parse(result.body), { request_errors: ["invalid instance_id"] });
+			assertNotRead(result);
+		});
+		it("should not read the body of an upload the host fails to receive", async function() {
+			this.timeout(30000);
+			controller.sendToHostByInstanceId = async (request) => {
+				throw new Error("Something went wrong");
+			};
+			for (const options of uploadVariants()) {
+				const result = await sendLargeBody(options.url, {
+					method: "POST",
+					headers: { ...options.headers, "X-Access-Token": testToken },
+					prefix: options.prefix,
+				});
+				assert.equal(result.status, 500);
+				assert.deepEqual(JSON.parse(result.body), { errors: ["Something went wrong"], request_errors: [] });
+				assertNotRead(result);
+			}
+		});
+		it("should not read the body of an upload the host never fetches", async function() {
+			this.timeout(30000);
+			controller.sendToHostByInstanceId = async (request) => {
+				await new Promise(() => {});
+			};
+			const result = await sendLargeBody(`${endpoint}?instance_id=123&filename=test.zip`, {
+				method: "POST", headers: { "Content-Type": "application/zip", "X-Access-Token": testToken },
+			});
+			assert.equal(result.status, 500);
+			assert.deepEqual(
+				JSON.parse(result.body), { errors: ["Timed out establishing stream to host"], request_errors: [] }
+			);
+			assertNotRead(result);
+		});
+		async function cutUploadShort(options) {
+			const request = http.request(options.url, {
+				method: "POST",
+				headers: { ...options.headers, "X-Access-Token": testToken, "Content-Length": 1024 * 1024 },
+			});
+			request.on("error", () => {});
+			const transfer = new Promise(resolve => {
+				controller.sendToHostByInstanceId = async (pullRequest) => {
+					const stream = await fetch(`http://localhost:${port}/api/stream/${pullRequest.streamId}`);
+					request.destroy();
+					try {
+						await stream.arrayBuffer();
+						resolve("completed");
+					} catch (err) {
+						resolve("failed");
+						throw err;
+					}
+					return pullRequest.name;
+				};
+			});
+			request.write(options.prefix ?? "");
+			request.write(Buffer.alloc(1024));
+			return await transfer;
+		}
+		it("should fail the transfer to the host when the upload is cut short", async function() {
+			for (const options of uploadVariants()) {
+				assert.equal(await cutUploadShort(options), "failed");
+			}
+		});
+	});
+	describe("/api/upload-mod", function() {
+		let endpoint, testToken, modsDirectory;
+		beforeEach(async function() {
+			endpoint = `http://localhost:${port}/api/upload-mod`;
+			testToken = controller.users.signUserToken({ id: "test" });
+			modsDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "clusterio-mods-"));
+			controller.mockConfigEntries.set("controller.mods_directory", modsDirectory);
+			controller.modStore = { addMod() {} };
+		});
+		afterEach(async function() {
+			await fs.rm(modsDirectory, { recursive: true, force: true });
+		});
+		async function waitForEmptyModsDirectory() {
+			for (let i = 0; i < 50 && (await fs.readdir(modsDirectory)).length; i++) {
+				await lib.wait(20);
+			}
+			assert.deepEqual(await fs.readdir(modsDirectory), []);
+		}
+
+		it("should respond with 403 when user has insufficient permission", async function() {
+			const user = controller.users.getOrCreateUser("player");
+			const response = await fetch(`${endpoint}?filename=mod_1.0.0.zip`, {
+				method: "POST",
+				headers: { "Content-Type": "application/zip", "X-Access-Token": controller.users.signUserToken(user) },
+				body: "totally a zip file",
+			});
+			assert.equal(response.status, 403);
+		});
+		it("should respond with 415 if content type is invalid", async function() {
+			const response = await fetch(`${endpoint}?filename=mod_1.0.0.zip`, {
+				method: "POST",
+				headers: { "Content-Type": "text/plain", "X-Access-Token": testToken },
+				body: "totally a zip file",
+			});
+			assert.equal(response.status, 415);
+			assert.deepEqual(await response.json(), { request_errors: ["invalid Content-Type"] });
+		});
+		it("should store an uploaded mod", async function() {
+			const info = await fs.readFile(path.join("test", "file", "empty_mod", "info.json"));
+			const zip = await createZipBuffer([["empty_mod_1.0.0/info.json", info]]);
+			const form = new FormData();
+			form.append("file", new Blob([zip], { type: "application/zip" }), "upload.zip");
+			for (const options of [
+				{ url: `${endpoint}?filename=upload.zip`, headers: { "Content-Type": "application/zip" }, body: zip },
+				{ url: endpoint, body: form },
+			]) {
+				const added = [];
+				controller.modStore = { addMod: mod => added.push(mod.filename) };
+				const response = await fetch(options.url, {
+					method: "POST", headers: { ...options.headers, "X-Access-Token": testToken }, body: options.body,
+				});
+				assert.equal(response.status, 200);
+				assert.equal((await response.json()).mods[0].name, "empty_mod");
+				assert.deepEqual(added, ["empty_mod_1.0.0.zip"]);
+				assert.deepEqual(await fs.readdir(modsDirectory), ["empty_mod_1.0.0.zip"]);
+			}
+		});
+		it("should respond with 500 and remove the file if it's not a mod", async function() {
+			for (const options of [
+				{ url: `${endpoint}?filename=mod_1.0.0.zip`, type: "application/zip", body: "totally a zip file" },
+				{ url: endpoint, body: new FormData() },
+			]) {
+				if (options.body instanceof FormData) {
+					options.body.append(
+						"file", new Blob(["totally a zip file"], { type: "application/zip" }), "mod_1.0.0.zip"
+					);
+				}
+				const response = await fetch(options.url, {
+					method: "POST",
+					headers: { ...options.type && { "Content-Type": options.type }, "X-Access-Token": testToken },
+					body: options.body,
+				});
+				assert.equal(response.status, 500);
+				const result = await response.json();
+				assert.equal(result.errors.length, 1);
+				assert.deepEqual(result.request_errors, []);
+				assert.deepEqual(await fs.readdir(modsDirectory), []);
+			}
+		});
+		it("should not read the body of an invalid file request", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(`${endpoint}?filename=mod_1.0.0.txt`, {
+				method: "POST", headers: { "Content-Type": "application/zip", "X-Access-Token": testToken },
+			});
+			assert.equal(result.status, 400);
+			assert.deepEqual(JSON.parse(result.body), { request_errors: ["filename must end with .zip"] });
+			assertNotRead(result);
+		});
+		it("should not read the body of an invalid multipart request", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(endpoint, {
+				method: "POST",
+				headers: { "Content-Type": "multipart/form-data; boundary=test", "X-Access-Token": testToken },
+				prefix: multipartPrefix("test", {}, "mod_1.0.0.zip", "text/plain"),
+			});
+			assert.equal(result.status, 400);
+			assert.deepEqual(JSON.parse(result.body), { request_errors: ["invalid file Content-Type"] });
+			assertNotRead(result);
+		});
+		it("should remove the file when the upload is cut short", async function() {
+			for (const options of [
+				{ url: `${endpoint}?filename=mod_1.0.0.zip`, headers: { "Content-Type": "application/zip" } },
+				{
+					url: endpoint,
+					headers: { "Content-Type": "multipart/form-data; boundary=test" },
+					prefix: multipartPrefix("test", {}, "mod_1.0.0.zip"),
+				},
+			]) {
+				const request = http.request(options.url, {
+					method: "POST",
+					headers: { ...options.headers, "X-Access-Token": testToken, "Content-Length": 1024 * 1024 },
+				});
+				request.on("error", () => {});
+				request.write(options.prefix ?? "");
+				request.write(Buffer.alloc(1024));
+				for (let i = 0; i < 50 && !(await fs.readdir(modsDirectory)).length; i++) {
+					await lib.wait(20);
+				}
+				assert.deepEqual(await fs.readdir(modsDirectory), ["mod_1.0.0.tmp.zip"]);
+				request.destroy();
+				await waitForEmptyModsDirectory();
+			}
+		});
+	});
+	describe("/api/upload-export", function() {
+		let endpoint;
+		beforeEach(function() {
+			endpoint = `http://localhost:${port}/api/upload-export`;
+		});
+		it("should not read the body of an upload without a valid token", async function() {
+			this.timeout(30000);
+			const result = await sendLargeBody(`${endpoint}?mod_pack_id=1`, {
+				headers: { "Content-Type": "application/zip" },
+			});
+			assert.equal(result.status, 401);
+			assertNotRead(result);
+		});
+		it("should not read the body of an invalid upload", async function() {
+			this.timeout(30000);
+			const hostToken = controller.hosts.signHostToken(controller.hosts.createHost("Host A").id);
+			const result = await sendLargeBody(`${endpoint}?mod_pack_id=invalid`, {
+				headers: { "Content-Type": "application/zip", "X-Access-Token": hostToken },
+			});
+			assert.equal(result.status, 400);
+			assertNotRead(result);
+		});
 	});
 });

test/mock.js

@@ -175,6 +175,7 @@ export class MockController {
 			["controller.public_url", "test"],
 			["controller.auth_secret", "TestSecretDoNotUse"],
 			["controller.proxy_stream_timeout", 1],
+			["controller.transfer_idle_timeout", 10],
 			["controller.metrics_timeout", 1],
 			["controller.host_pairing_timeout", 3600],
 			["controller.host_pairing_max_pending", 50],