diff --git a/docs/configuration.md b/docs/configuration.md index 2bc2e221..009f54bd 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -241,6 +241,19 @@ Must be greater than 0. Defaults to 15. +### controller.transfer_min_progress + +Amount in KiB a request to the controller has to move every 30 seconds, requests moving less are closed. +This closes uploads and downloads that have stalled or are sent too slowly to ever finish, while there is no limit on how long a transfer making progress can take. +It applies to saves, mods and exports uploaded to the controller, proxy streams through it, and the body of any other request. +Time spent waiting for a proxy stream to start is not counted, see controller.proxy_stream_timeout. +Data is passed on in chunks of up to 64 KiB and progress is counted per chunk, so a proxy stream moving less than about that in 30 seconds may also be closed. +Set to 0 for no limit. +Must be 0 or greater. + +Defaults to 30. + + ### controller.factorio_username Username of the Factorio.com account to authenticate with the Factorio API with. diff --git a/docs/setting-up-tls.md b/docs/setting-up-tls.md index a063b868..89073068 100644 --- a/docs/setting-up-tls.md +++ b/docs/setting-up-tls.md @@ -19,6 +19,12 @@ Additianlly WebSocket connections to `/api/socket` also needs to be forwarded. Typically this requires explicit configuration and support for forwarding WebSocket connections. If the interface hosted under a sub-path the `/api/socket` address is relative to the sub-path and the sub-path also needs to be stripped out when forwarded. +Saves are sent from hosts to the web interface through the controller as they are uploaded to `/api/stream/`. +If the proxy buffers request bodies it holds back a host's upload until all of it has arrived. +A save that takes longer than `controller.proxy_stream_timeout` to upload then fails, after the whole save was sent. +Turn request buffering off for the proxy, with Nginx this is `proxy_request_buffering off;`. +Nginx also limits request bodies to 1 MB by default, set `client_max_body_size` high enough for the saves and mods you upload. + You can also optionally set the controller to listen only on the loopback interface to prevent it being directly connected to from the outside. npx clusteriocontroller config set controller.bind_address 127.0.0.1 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 2d2a36ca..9c5121d5 100644 --- a/packages/controller/src/routes.ts +++ b/packages/controller/src/routes.ts @@ -417,6 +417,9 @@ export interface ProxyStream { filename: string | null; events: events.EventEmitter; timeout: ReturnType; + closed: boolean; + /** Bytes the receiver has taken, set once the stream is flowing */ + progress?: () => number; } export async function createProxyStream(app: Application): Promise { @@ -433,8 +436,10 @@ export async function createProxyStream(app: Application): Promise timeout: setTimeout(() => { stream.events.emit("timeout"); }, app.locals.controller.config.get("controller.proxy_stream_timeout") * 1000), + closed: false, }; stream.events.on("close", () => { + stream.closed = true; clearTimeout(stream.timeout); app.locals.streams.delete(id); }); @@ -445,24 +450,105 @@ export async function createProxyStream(app: Application): Promise return stream; } +/** + * 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. + * Requests are instead closed by guardRequest when they move too little. + * Turning the limit off also turns off the limit on receiving the headers, + * which guardRequest can't cover, so that is set back to its default. + */ +export const httpServerOptions = { requestTimeout: 0, headersTimeout: 60_000 }; + +/** Time an upload is read and thrown away after it was answered, before its connection is closed */ +const answeredUploadDrainMs = 1000; + +/** + * Guard requests 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. + * + * The connection is also closed if the request moved less than + * controller.transfer_min_progress during a check interval, so that a + * request which stalled or is trickled in does not stay open forever while + * a slow one can take the time it needs. For a request body this is what + * was received of it. For both ends of a proxy stream it is what the + * receiver has taken, as the sender is held back when the receiver is the + * slow one. Time spent waiting for a proxy stream to start is not counted, + * controller.proxy_stream_timeout limits that. + * @param checkIntervalMs - Interval to check the progress of requests on. + * @returns Express middleware. + */ +function guardRequest(checkIntervalMs: number) { + return function(req: Request, res: Response, next: any) { + const socket = req.socket; + 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); + }); + }); + + const received = () => socket.bytesRead; + let measure: (() => number) | undefined = received; + let last = received(); + const check = setInterval(() => { + const stream: ProxyStream | undefined = res.locals.proxyStream; + let current: (() => number) | undefined; + if (stream && !stream.closed) { + current = stream.progress; + } else if (!req.complete) { + current = received; + } + const minProgress = req.app.locals.controller.config.get("controller.transfer_min_progress"); + if (current !== measure || !current || !minProgress) { + // What to measure changed, start over from here. + measure = current; + last = current ? current() : 0; + return; + } + const progress = current(); + if (progress - last < minProgress * 1024) { + logger.warn( + `Closing ${req.method} ${req.path}, it moved less than ${minProgress} KiB ` + + `in ${checkIntervalMs / 1000} seconds` + ); + socket.destroy(); + } + last = progress; + }, checkIntervalMs); + res.once("close", () => clearInterval(check)); + next(); + }; +} + async function putStream(req: Request<{ id: string }>, res: Response) { const stream = req.app.locals.streams.get(req.params.id); if (!stream || stream.source) { res.sendStatus(404); return; } + res.locals.proxyStream = stream; 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); - } + // The stream also closes when it timed out or the receiver disconnected. + res.sendStatus(stream.flowing && req.complete ? 200 : 500); }); } @@ -476,9 +562,22 @@ function startStream(res: Response, stream: ProxyStream) { } else { res.append("Content-Disposition", "attachment"); } - stream.source!.pipe(res); + const source = stream.source!; + const socket = res.socket!; + source.pipe(res); stream.flowing = true; + // Counts completed writes, so data waiting in a buffer is not progress. + stream.progress = () => socket.bytesWritten - socket.writableLength; + // 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); @@ -490,6 +589,7 @@ async function getStream(req: Request<{ id: string }>, res: Response) { res.sendStatus(404); return; } + res.locals.proxyStream = stream; if (stream.source) { startStream(res, stream); @@ -509,6 +609,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"); @@ -517,26 +752,25 @@ 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; + res.locals.proxyStream = proxyStream; + 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")); }); }); @@ -547,7 +781,7 @@ async function uploadSave(req: Request, res: Response) { new lib.InstancePullSaveRequest( instanceId, proxyStream.id, - filename, + file.filename, ) ), timeout, @@ -556,90 +790,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) { @@ -658,37 +815,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); @@ -697,11 +835,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 { @@ -711,76 +846,18 @@ 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 }); } -export function addRouteHandlers(app: Application) { +export function addRouteHandlers(app: Application, progressCheckIntervalMs = 30_000) { + app.use(guardRequest(progressCheckIntervalMs)); app.get("/metrics", validateUserToken, (req:Request, res:Response, next:any) => getMetrics(req, res).catch(next) diff --git a/packages/lib/src/config/definitions.ts b/packages/lib/src/config/definitions.ts index ecea3400..86dd1b84 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_min_progress": number; "controller.factorio_username": string | null; "controller.factorio_token": string | null; "controller.share_factorio_credentials_with_hosts": boolean; @@ -223,6 +224,13 @@ export class ControllerConfig extends classes.Config { initialValue: 15, validator: validators.greaterThanZero, }, + "controller.transfer_min_progress": { + title: "Transfer Minimum Progress", + description: "KiB an upload or download has to move every 30 seconds to not be closed, 0 for no limit.", + type: "number", + initialValue: 30, + validator: validators.greaterThanEqualZero, + }, "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 50d81986..32669064 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"; @@ -21,8 +22,9 @@ describe("controller/src/routes", function() { let port; beforeEach(async function() { controller = new mock.MockController(); - routes.addRouteHandlers(controller.app); - server = http.createServer(controller.app); + // Check the progress of requests every 500 ms instead of every 30 seconds. + routes.addRouteHandlers(controller.app, 500); + server = http.createServer(routes.httpServerOptions, controller.app); server.listen(0, "localhost"); await events.once(server, "listening"); port = server.address().port; @@ -30,6 +32,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,6 +112,141 @@ describe("controller/src/routes", function() { let response = await fetch(`${endpoint}/${stream.id}`); assert.equal(response.status, 500); }); + it("should not read the body of a PUT to a stream that doesn't exist", async function() { + this.timeout(30000); + const result = await sendLargeBody(`${endpoint}/01020304`); + assert.equal(result.status, 404); + assertNotRead(result); + }); + it("should not read the body of a PUT to a stream that times out", async function() { + this.timeout(30000); + const stream = await routes.createProxyStream(controller.app); + const result = await sendLargeBody(`${endpoint}/${stream.id}`); + assert.equal(result.status, 500); + 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()); + }); + /** + * PUT to a stream at a steady pace until the connection closes or the body is sent + * @param {string} id - Stream to send to. + * @param {number} chunkSize - Bytes to send every 100 ms. + * @param {number} chunks - Number of chunks the body consists of. + * @returns {Promise} status responded with, if any. + */ + async function putPaced(id, chunkSize, chunks) { + const request = http.request( + `${endpoint}/${id}`, { method: "PUT", headers: { "Content-Length": chunkSize * chunks } } + ); + let status; + request.on("response", response => { + status = response.statusCode; + response.resume(); + }); + request.on("error", () => {}); + const closed = new Promise(resolve => request.on("close", resolve)); + for (let i = 0; i < chunks && !request.closed; i++) { + request.write(Buffer.alloc(chunkSize)); + await lib.wait(100); + } + if (!request.closed) { + request.end(); + } + await closed; + return status; + } + it("should close a transfer that stops making progress", async function() { + controller.mockConfigEntries.set("controller.transfer_min_progress", 1); + 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 close a transfer that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + const stream = await routes.createProxyStream(controller.app); + const start = Date.now(); + const upload = putPaced(stream.id, 200, 50); + const response = await fetch(`${endpoint}/${stream.id}`); + assert.equal(response.status, 200); + await assert.rejects(response.arrayBuffer()); + assert.equal(await upload, undefined); + assert(Date.now() - start < 2000, `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_min_progress", 1); + const stream = await routes.createProxyStream(controller.app); + const upload = putPaced(stream.id, 1024, 20); + const download = fetch(`${endpoint}/${stream.id}`).then(response => response.arrayBuffer()); + assert.equal(await upload, 200); + assert.equal((await download).byteLength, 20 * 1024); + }); + it("should not close a transfer held back by a slow receiver", async function() { + this.timeout(20000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 16); + const size = 4 * 1024 * 1024; + const stream = await routes.createProxyStream(controller.app); + const upload = fetch(`${endpoint}/${stream.id}`, { method: "PUT", body: Buffer.alloc(size) }); + const received = await new Promise((resolve, reject) => { + http.get(`${endpoint}/${stream.id}`, response => { + let bytes = 0; + response.on("data", chunk => { + bytes += chunk.length; + response.pause(); + setTimeout(() => response.resume(), 50); + }); + response.on("end", () => resolve(bytes)); + response.on("error", reject); + }).on("error", reject); + }); + assert.equal(received, size); + assert.equal((await upload).status, 200); + }); + it("should not count the time waiting for the stream to start", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.proxy_stream_timeout", 3); + controller.mockConfigEntries.set("controller.transfer_min_progress", 1); + const stream = await routes.createProxyStream(controller.app); + const download = fetch(`${endpoint}/${stream.id}`); + await lib.wait(1500); + const upload = await fetch(`${endpoint}/${stream.id}`, { method: "PUT", body: "test content" }); + assert.equal(upload.status, 200); + assert.equal(await (await download).text(), "test content"); + }); it("should passthrough a stream", async function() { let stream; let responses; @@ -808,5 +1006,312 @@ 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("guardRequest()", function() { + async function trickle(url, headers) { + const request = http.request(url, { method: "POST", headers: { ...headers, "Content-Length": 2 ** 20 } }); + let status; + request.on("response", response => { + status = response.statusCode; + response.resume(); + }); + request.on("error", () => {}); + const closed = new Promise(resolve => request.on("close", resolve)); + const start = Date.now(); + while (!request.closed && Date.now() - start < 5000) { + request.write(Buffer.alloc(200)); + await lib.wait(100); + } + request.destroy(); + await closed; + return { status, duration: Date.now() - start }; + } + it("should keep the limit on receiving headers", function() { + assert.equal(server.requestTimeout, 0); + assert.equal(server.headersTimeout, 60000); + }); + it("should close an upload that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + const modsDirectory = await fs.mkdtemp(path.join(os.tmpdir(), "clusterio-mods-")); + try { + controller.mockConfigEntries.set("controller.mods_directory", modsDirectory); + const result = await trickle(`http://localhost:${port}/api/upload-mod?filename=mod_1.0.0.zip`, { + "Content-Type": "application/zip", + "X-Access-Token": controller.users.signUserToken({ id: "test" }), + }); + assert.equal(result.status, undefined); + assert(result.duration < 2000, `closed after ${result.duration} ms`); + for (let i = 0; i < 50 && (await fs.readdir(modsDirectory)).length; i++) { + await lib.wait(20); + } + assert.deepEqual(await fs.readdir(modsDirectory), []); + } finally { + await fs.rm(modsDirectory, { recursive: true, force: true }); + } + }); + it("should close a request to another route that is trickled in", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 4); + controller.app.post("/test", express.raw({ type: "*/*", limit: "2mb" }), (req, res) => res.sendStatus(200)); + const result = await trickle(`http://localhost:${port}/test`, { "Content-Type": "text/plain" }); + assert.equal(result.status, undefined); + assert(result.duration < 2000, `closed after ${result.duration} ms`); + }); + it("should not close a trickled in request when the limit is 0", async function() { + this.timeout(10000); + controller.mockConfigEntries.set("controller.transfer_min_progress", 0); + controller.app.post("/test", express.raw({ type: "*/*", limit: "2mb" }), (req, res) => res.sendStatus(200)); + const result = await trickle(`http://localhost:${port}/test`, { "Content-Type": "text/plain" }); + assert(result.duration >= 5000, `closed after ${result.duration} ms`); + }); + }); + 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..32b5270b 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_min_progress", 30], ["controller.metrics_timeout", 1], ["controller.host_pairing_timeout", 3600], ["controller.host_pairing_max_pending", 50],