feat(upload): abort clamd + S3 streams on client disconnect

Propagate request.signal end-to-end through createClamdStream,
createMultipartUpload, handleStreamingUpload and runUploadPipeline.
Combine with AbortSignal.timeout(UPLOAD_REQUEST_TIMEOUT_MS = 120s) via
AbortSignal.any so a hung scan or S3 socket cannot tie up the handler
indefinitely.

On abort the socket is destroyed, the multipart upload is aborted and
the reader is cancelled. Also normalises the file extension spliced
into the S3 key to alnum only (defence-in-depth on top of path.extname).

Author: maxgfr

packages/app/src/modules/shared/uploadConfig.ts

@@ -17,6 +17,13 @@ export const ALLOWED_UPLOAD_MIME_TYPES = [
 /** ClamAV scan timeout in milliseconds (30s). */
 export const SCAN_TIMEOUT_MS = 30_000;
 
+/**
+ * Overall upload pipeline timeout (2min). Combined via AbortSignal.any with
+ * the incoming request signal so a hung clamd or S3 socket aborts the whole
+ * pipeline instead of tying up the request indefinitely.
+ */
+export const UPLOAD_REQUEST_TIMEOUT_MS = 120_000;
+
 /** Minimum S3 multipart part size in bytes (5 MB, required by S3 except for the last part). */
 export const S3_PART_MIN_SIZE = 5 * 1024 * 1024;
 

packages/app/src/server/services/__tests__/clamav.test.ts

@@ -155,6 +155,34 @@ describe("clamav service", () => {
 		expect(socket.destroy).toHaveBeenCalled();
 	});
 
+	it("destroys the clamd socket immediately when the abort signal fires", async () => {
+		// Fresh socket for this test.
+		net.createConnection({ host: "x", port: 0 });
+		const socket = getSocket();
+		socket.on.mockImplementation(() => socket);
+
+		const { createClamdStream } = await import("../clamav");
+		const controller = new AbortController();
+		createClamdStream("localhost", 3310, controller.signal);
+
+		socket.destroy.mockClear();
+		controller.abort();
+		expect(socket.destroy).toHaveBeenCalled();
+	});
+
+	it("rejects sendChunk if the abort signal is already aborted at construction", async () => {
+		net.createConnection({ host: "x", port: 0 });
+		const socket = getSocket();
+		socket.on.mockImplementation(() => socket);
+
+		const { createClamdStream } = await import("../clamav");
+		const controller = new AbortController();
+		controller.abort();
+		const clamd = createClamdStream("localhost", 3310, controller.signal);
+
+		await expect(clamd.sendChunk(Buffer.from("x"))).rejects.toThrow(/aborted/i);
+	});
+
 	it("surfaces a connection error thrown by the socket during sendChunk", async () => {
 		// Trigger `createConnection` once to populate the mock's results
 		// array, then grab the shared socket instance before `createClamdStream`

packages/app/src/server/services/clamav.ts

@@ -16,15 +16,37 @@ export type ScanResult = { clean: true } | { clean: false; virus: string };
  *   → [4 bytes big-endian = chunk size][chunk data]  (repeated)
  *   → [4 bytes zero]                                 (end of stream)
  *   ← "stream: OK\0"  or  "stream: <virus> FOUND\0"
+ *
+ * When an `AbortSignal` is provided, an abort (client disconnect, request
+ * timeout) immediately destroys the socket. Any pending `sendChunk` or
+ * `finish` call rejects with the abort reason so the pipeline surfaces a
+ * clean failure instead of hanging on the clamd read.
  */
-export function createClamdStream(host: string, port: number) {
+export function createClamdStream(
+	host: string,
+	port: number,
+	signal?: AbortSignal,
+) {
 	const socket = net.createConnection({ host, port });
 
 	let socketError: Error | null = null;
 	socket.on("error", (err) => {
 		socketError = err;
 	});
 
+	const onAbort = () => {
+		socketError =
+			socketError ?? new Error("clamd connection aborted by request signal");
+		socket.destroy();
+	};
+	if (signal) {
+		if (signal.aborted) {
+			onAbort();
+		} else {
+			signal.addEventListener("abort", onAbort, { once: true });
+		}
+	}
+
 	// Send the INSTREAM command (null-terminated)
 	socket.write("zINSTREAM\0");
 
@@ -97,8 +119,11 @@ export function createClamdStream(host: string, port: number) {
 /**
  * Convenience wrapper: scan a complete buffer using the INSTREAM protocol.
  */
-export async function scanBuffer(buffer: Buffer): Promise<ScanResult> {
-	const clamd = createClamdStream(env.CLAMAV_HOST, env.CLAMAV_PORT);
+export async function scanBuffer(
+	buffer: Buffer,
+	signal?: AbortSignal,
+): Promise<ScanResult> {
+	const clamd = createClamdStream(env.CLAMAV_HOST, env.CLAMAV_PORT, signal);
 
 	try {
 		await clamd.sendChunk(buffer);

packages/app/src/server/services/fileUpload.ts

@@ -22,6 +22,11 @@ type UploadOptions = {
 	maxSize?: number;
 	/** MIME types to allow, validated via magic bytes on the first chunk. */
 	allowedMimeTypes: readonly string[];
+	/**
+	 * Request / timeout signal. On abort, the clamd socket is destroyed and
+	 * the S3 multipart is aborted, so no orphan parts are left behind.
+	 */
+	signal?: AbortSignal;
 };
 
 /**
@@ -36,7 +41,8 @@ export type UploadFailureReason =
 	| "wrong_type"
 	| "empty"
 	| "virus"
-	| "scan_unavailable";
+	| "scan_unavailable"
+	| "aborted";
 
 export type UploadResult =
 	| { ok: true; key: string; fileId: string }
@@ -60,12 +66,16 @@ export async function handleStreamingUpload(
 	options: UploadOptions,
 ): Promise<UploadResult> {
 	const maxSize = options.maxSize ?? MAX_FILE_SIZE;
-	const ext = path.extname(options.fileName) || ".bin";
+	const signal = options.signal;
+	// Normalise ext to alnum only — `path.extname` rules out `/` traversal
+	// (basename suffix) but exotic bytes would survive and land in the S3 key.
+	const rawExt = path.extname(options.fileName);
+	const ext = /^\.[A-Za-z0-9]{1,10}$/.test(rawExt) ? rawExt : ".bin";
 	const fileId = crypto.randomUUID();
 	const key = `${options.siren}/${options.year}/${fileId}${ext}`;
 
-	const clamd = createClamdStream(env.CLAMAV_HOST, env.CLAMAV_PORT);
-	const s3Upload = createMultipartUpload(key, options.contentType);
+	const clamd = createClamdStream(env.CLAMAV_HOST, env.CLAMAV_PORT, signal);
+	const s3Upload = createMultipartUpload(key, options.contentType, signal);
 	await s3Upload.init();
 
 	let totalBytes = 0;
@@ -74,6 +84,7 @@ export async function handleStreamingUpload(
 
 	try {
 		while (true) {
+			if (signal?.aborted) throw new UploadAbortedError();
 			const { done, value } = await reader.read();
 			if (done) break;
 
@@ -103,19 +114,29 @@ export async function handleStreamingUpload(
 				s3Upload.sendChunk(buf),
 			]);
 			if (clamdResult.status === "rejected") {
+				if (signal?.aborted) throw new UploadAbortedError();
 				throw new ClamdScanError(clamdResult.reason);
 			}
 			if (s3Result.status === "rejected") {
+				if (signal?.aborted) throw new UploadAbortedError();
 				throw s3Result.reason;
 			}
 		}
 	} catch (err) {
 		clamd.destroy();
 		await s3Upload.abort().catch(() => {});
+		await reader.cancel().catch(() => {});
 
 		if (err instanceof FileTooLargeError) {
 			return { ok: false, reason: "too_large", error: FILE_TOO_LARGE_ERROR };
 		}
+		if (err instanceof UploadAbortedError || signal?.aborted) {
+			return {
+				ok: false,
+				reason: "aborted",
+				error: "L'upload a été interrompu.",
+			};
+		}
 		if (err instanceof ClamdScanError) {
 			console.error("[fileUpload] clamd sendChunk failed", err.cause);
 			return {
@@ -144,6 +165,13 @@ export async function handleStreamingUpload(
 		console.error("[fileUpload] clamd scan failed", err);
 		clamd.destroy();
 		await s3Upload.abort().catch(() => {});
+		if (signal?.aborted) {
+			return {
+				ok: false,
+				reason: "aborted",
+				error: "L'upload a été interrompu.",
+			};
+		}
 		return {
 			ok: false,
 			reason: "scan_unavailable",
@@ -180,6 +208,17 @@ class FileTooLargeError extends Error {
 	}
 }
 
+/**
+ * Sentinel thrown when the request-scoped AbortSignal fires (client
+ * disconnect or request-level timeout). Lets the outer catch distinguish an
+ * interruption from a genuine scan/S3 failure.
+ */
+class UploadAbortedError extends Error {
+	constructor() {
+		super("Upload aborted by request signal");
+	}
+}
+
 /**
  * Wraps a failure from `clamd.sendChunk()` so the outer catch can distinguish
  * an antivirus-availability issue from a generic I/O failure and surface it

packages/app/src/server/services/s3.ts

@@ -117,8 +117,17 @@ import { S3_PART_MIN_SIZE } from "~/modules/shared/uploadConfig";
 /**
  * Creates an incremental multipart upload to S3.
  * Chunks are buffered until they reach S3_PART_MIN_SIZE, then flushed as a part.
+ *
+ * The optional `signal` is forwarded to every `s3Client.send()` call so an
+ * aborted request (client disconnect, request-level timeout) cancels the
+ * in-flight HTTP request at the AWS SDK layer. Calling `abort()` is still
+ * required on the caller side to release the multipart upload on S3.
  */
-export function createMultipartUpload(key: string, contentType: string) {
+export function createMultipartUpload(
+	key: string,
+	contentType: string,
+	signal?: AbortSignal,
+) {
 	let uploadId: string;
 	let partNumber = 1;
 	const parts: { ETag: string; PartNumber: number }[] = [];
@@ -132,6 +141,7 @@ export function createMultipartUpload(key: string, contentType: string) {
 					Key: key,
 					ContentType: contentType,
 				}),
+				{ abortSignal: signal },
 			);
 			if (!res.UploadId) {
 				throw new Error("S3 CreateMultipartUpload returned no UploadId");
@@ -157,6 +167,7 @@ export function createMultipartUpload(key: string, contentType: string) {
 					PartNumber: partNumber,
 					Body: buffer,
 				}),
+				{ abortSignal: signal },
 			);
 			if (!res.ETag) {
 				throw new Error("S3 UploadPart returned no ETag");
@@ -175,10 +186,15 @@ export function createMultipartUpload(key: string, contentType: string) {
 					UploadId: uploadId,
 					MultipartUpload: { Parts: parts },
 				}),
+				{ abortSignal: signal },
 			);
 		},
 
 		async abort() {
+			// Intentionally not forwarding `signal` here: abort() runs in the
+			// catch/cleanup path where the outer signal is usually already
+			// aborted, which would itself short-circuit the AbortMultipartUpload
+			// call and leave the upload alive on S3.
 			return s3Client.send(
 				new AbortMultipartUploadCommand({
 					Bucket: env.S3_BUCKET_NAME,

packages/app/src/server/services/uploadPipeline.ts

@@ -6,6 +6,7 @@ import { MAX_CSE_FILES } from "~/modules/cseOpinion/types";
 import {
 	ALLOWED_UPLOAD_MIME_TYPES,
 	type FlowType,
+	UPLOAD_REQUEST_TIMEOUT_MS,
 } from "~/modules/shared/uploadConfig";
 import { db } from "~/server/db";
 import { declarations, files } from "~/server/db/schema";
@@ -50,6 +51,12 @@ export type UploadPipelineInput = {
 	contentType: string;
 	stream: ReadableStream<Uint8Array>;
 	flowType: FlowType;
+	/**
+	 * Abort signal from the HTTP request. Combined with a pipeline-level
+	 * timeout and forwarded to clamd + S3 so a client disconnect or hung
+	 * socket aborts the whole stream instead of tying up the request.
+	 */
+	signal?: AbortSignal;
 };
 
 /**
@@ -101,12 +108,15 @@ export async function runUploadPipeline(
 		}
 	}
 
+	const pipelineSignal = combineSignals(input.signal);
+
 	const streamResult = await handleStreamingUpload(input.stream, {
 		siren: input.siren,
 		year: input.year,
 		fileName: input.fileName,
 		contentType: input.contentType,
 		allowedMimeTypes: ALLOWED_UPLOAD_MIME_TYPES,
+		signal: pipelineSignal,
 	});
 
 	if (!streamResult.ok) {
@@ -284,3 +294,14 @@ export class MaxFilesReachedError extends Error {
 		super("Max files reached (race condition)");
 	}
 }
+
+/**
+ * Combine the incoming request signal with a pipeline-level timeout. Either
+ * one firing aborts the clamd + S3 streams. Requires Node 20+ (native
+ * AbortSignal.any + AbortSignal.timeout).
+ */
+function combineSignals(requestSignal: AbortSignal | undefined): AbortSignal {
+	const timeoutSignal = AbortSignal.timeout(UPLOAD_REQUEST_TIMEOUT_MS);
+	if (!requestSignal) return timeoutSignal;
+	return AbortSignal.any([requestSignal, timeoutSignal]);
+}