mirror of
https://github.com/clusterio/clusterio.git
synced 2026-11-14 07:24:52 +09:00
Stop reading rejected save uploads to the controller (#1173)
* Stop reading rejected save uploads to the controller When a host pushes a save to a stream that is gone, or that timed out before anyone downloaded it, the controller answers 404 or 500 without reading the body. Node then reads and throws away the rest of the body to keep the connection open, so a proxy in front of the controller that keeps passing the upload on made the whole save get uploaded for nothing. Close the connection if the body hasn't ended a second after the response. Document turning off request buffering in reverse proxies, as a buffering proxy holds the upload back until all of it has arrived. * Remove the close listener when a rejected upload ends A connection kept open for more requests got one more close listener for every rejected upload that finished within the grace period. * 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. * Close requests moving too little instead of idle ones Replace controller.transfer_idle_timeout with controller.transfer_min_progress, the amount a request has to move every 30 seconds, so a request trickled in is closed too. The check now runs for every route. Both ends of a proxy stream are judged on what the receiver has taken, so a sender held back by a slow receiver is not closed. Keep the limit on receiving headers, which turning off the request timeout also turned off.
This commit is contained in:
7 files changed
+809
-198
No files matched your search
@@ -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.
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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);
|
||||
|
||||
+272
-195
@@ -417,6 +417,9 @@ export interface ProxyStream {
|
||||
filename: string | null;
|
||||
events: events.EventEmitter<ProxyStreamEvents>;
|
||||
timeout: ReturnType<typeof setTimeout>;
|
||||
closed: boolean;
|
||||
/** Bytes the receiver has taken, set once the stream is flowing */
|
||||
progress?: () => number;
|
||||
}
|
||||
|
||||
export async function createProxyStream(app: Application): Promise<ProxyStream> {
|
||||
@@ -433,8 +436,10 @@ export async function createProxyStream(app: Application): Promise<ProxyStream>
|
||||
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<ProxyStream>
|
||||
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<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");
|
||||
@@ -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<never>((_, 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)
|
||||
|
||||
@@ -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<ControllerConfigFields> {
|
||||
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.",
|
||||
|
||||
+507
-2
@@ -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<number|undefined>} 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);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -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],
|
||||
|
||||
Reference in new issue
Block a user