diff --git a/docs/configuration.md b/docs/configuration.md index 803e51f8..f87f148e 100644 --- a/docs/configuration.md +++ b/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. diff --git a/packages/controller/src/Controller.ts b/packages/controller/src/Controller.ts index e5058ff2..7b0ef1fa 100644 --- a/packages/controller/src/Controller.ts +++ b/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); diff --git a/packages/controller/src/routes.ts b/packages/controller/src/routes.ts index 8d7518a7..c61fe9b0 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -445,41 +445,62 @@ export async function createProxyStream(app: Application): Promise 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; + /** 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, +): Promise { + 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(); + 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(resolve => { + const fields = new Map(); + const tasks: Promise[] = []; + 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((_, 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) ); diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index 5fd9e64f..06a865ce 100644 --- a/packages/lib/src/config/definitions.ts +++ b/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 { 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.", diff --git a/test/controller/routes.js b/test/controller/routes.js index b99f8507..dd764e68 100644 --- a/test/controller/routes.js +++ b/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); + }); }); }); diff --git a/test/mock.js b/test/mock.js index 57b428a7..b64e0b0b 100644 --- a/test/mock.js +++ b/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],