From fef428ebea034a1a4da2267a0713e1124790c1fc Mon Sep 17 00:00:00 2001 From: Sivert Date: Mon, 5 Oct 2026 10:49:07 +0200 Subject: [PATCH 1/2] Send avatars, banners and chat images through the worker's quarantine (GRYT-1664) When the image worker reports quarantine-v1 on /health, an avatar, a banner or a chat image is stored under quarantine/ as sent, with an image job, and the server never decodes it: the worker writes it out again inside its jail and moves the row. Any read of a file still in quarantine waits up to 15 seconds for that, then answers 503 media_not_ready with no-store, so the bytes as sent are never served. A worker that doesn't say quarantine-v1, including none at all and the desktop-embedded one, keeps today's path. Videos, group and webhook pictures and emoji are unchanged for now. Redoes the reverted 8d7f9bd, gated on the worker. Co-Authored-By: Claude Opus 5.5 --- src/index.ts | 2 + src/routes/uploadQuarantine.test.ts | 120 ++++++++++++++++++++++++++++ src/routes/uploads.ts | 28 ++++++- src/services/quarantineUpload.ts | 58 ++++++++++++++ src/services/storeUploadedFile.ts | 10 ++- src/services/workerCapabilities.ts | 32 ++++++++ 6 files changed, 245 insertions(+), 5 deletions(-) create mode 100644 src/routes/uploadQuarantine.test.ts create mode 100644 src/services/quarantineUpload.ts create mode 100644 src/services/workerCapabilities.ts diff --git a/src/index.ts b/src/index.ts index 53a680d5..e0b906d9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,5 @@ import { rm } from "fs/promises"; +import { startWorkerCapabilityPolling } from "./services/workerCapabilities"; import { join } from "path"; import { config } from "dotenv"; import { applyCors, isOriginAllowed, readAllowedOrigins } from "./config/cors"; @@ -486,6 +487,7 @@ function reachableAddresses(port: number, host: string): string[] { } httpServer.listen(PORT, HOST, () => { + startWorkerCapabilityPolling(); consola.box(`Gryt Server v${VERSION}`); consola.start(`Starting ${process.env.SERVER_NAME}...`); if (process.env.SFU_WS_HOST) diff --git a/src/routes/uploadQuarantine.test.ts b/src/routes/uploadQuarantine.test.ts new file mode 100644 index 00000000..9edc153c --- /dev/null +++ b/src/routes/uploadQuarantine.test.ts @@ -0,0 +1,120 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, rmSync } from "node:fs"; +import type { AddressInfo } from "node:net"; +import type { Server } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import express from "express"; +import sharp from "sharp"; + +import { getSqliteDb, initSqlite } from "../db/sqlite/connection"; +import { getFile } from "../db/sqlite/messages"; +import { createServerConfigIfNotExists, setServerRole } from "../db/sqlite/servers"; +import { getUserByServerId, upsertUser } from "../db/sqlite/users"; +import { refreshWorkerCapabilities } from "../services/workerCapabilities"; +import { initStorage, putObject } from "../storage"; +import { generateAccessToken, generateFileToken } from "../utils/jwt"; + +/* GRYT-1664: with a worker that clears quarantine, an avatar or banner is stored out of reach + and nobody is ever served the bytes as sent; a read waits for the worker's copy instead. */ + +process.env.GRYT_QUARANTINE_WAIT_MS = "1500"; +// eslint-disable-next-line @typescript-eslint/no-require-imports +const { uploadsRouter } = require("./uploads") as typeof import("./uploads"); + +let dir: string; +let server: Server; +let base = ""; +let host = ""; + +interface Member { serverUserId: string; grytUserId: string; nickname: string } + +async function member(nickname: string): Promise { + const grytUserId = `acct-quarantine-${nickname}`; + const user = await upsertUser(grytUserId, nickname); + await setServerRole(user.server_user_id, "member"); + return { serverUserId: user.server_user_id, grytUserId, nickname }; +} + +const auth = (who: Member) => ({ Authorization: `Bearer ${generateAccessToken({ ...who, serverHost: host, tokenVersion: 0 })}` }); +const png = () => sharp({ create: { width: 800, height: 600, channels: 3, background: "#c83c5a" } }).png().toBuffer(); +const workerSays = (capabilities: string[]) => + refreshWorkerCapabilities("http://worker.test", (async () => new Response(JSON.stringify({ capabilities }))) as typeof fetch); + +async function upload(who: Member, path: "banner" | "avatar") { + const form = new FormData(); + form.append("file", new Blob([new Uint8Array(await png())], { type: "image/png" }), "picture.png"); + const res = await fetch(`${base}/api/uploads/${path}`, { method: "POST", headers: auth(who), body: form }); + return { status: res.status, body: (await res.json()) as Record }; +} + +const read = (who: Member, fileId: string) => + fetch(`${base}/api/uploads/files/${fileId}?t=${encodeURIComponent(generateFileToken({ ...who, serverHost: host, tokenVersion: 0 }))}`); + +before(async () => { + dir = mkdtempSync(join(tmpdir(), "gryt-quarantine-")); + process.env.DATA_DIR = dir; + process.env.S3_BUCKET = "gryt-test"; + process.env.STORAGE_BACKEND = "filesystem"; + await initSqlite(); + initStorage(); + await createServerConfigIfNotExists(); + const app = express(); + app.use("/api/uploads", uploadsRouter); + server = app.listen(0, "127.0.0.1"); + await new Promise((resolve) => server.once("listening", () => resolve())); + host = `127.0.0.1:${(server.address() as AddressInfo).port}`; + base = `http://${host}`; +}); + +after(async () => { + await new Promise((resolve) => server.close(() => resolve())); + rmSync(dir, { recursive: true, force: true }); +}); + +describe("uploads with a worker that clears quarantine", () => { + it("stores a banner out of reach and queues it for the worker", async () => { + assert.equal(await workerSays(["quarantine-v1"]), true); + const alice = await member("alice"); + const res = await upload(alice, "banner"); + assert.equal(res.status, 201, JSON.stringify(res.body)); + const fileId = res.body.bannerFileId as string; + assert.equal(res.body.processing, true); + assert.equal((await getUserByServerId(alice.serverUserId))?.banner_file_id, fileId); + const file = await getFile(fileId); + assert.ok(file?.s3_key.startsWith("quarantine/banners/"), file?.s3_key); + const job = getSqliteDb().prepare("SELECT raw_s3_key, status FROM image_jobs WHERE file_id = ?").get(fileId) as { raw_s3_key: string; status: string }; + assert.deepEqual({ ...job }, { raw_s3_key: file!.s3_key, status: "queued" }); + }); + + it("never serves the original: a read waits for the worker's copy, and says not ready if it never comes", async () => { + await workerSays(["quarantine-v1"]); + const bob = await member("bob"); + const fileId = (await upload(bob, "avatar")).body.avatarFileId as string; + assert.ok((await getFile(fileId))?.s3_key.startsWith("quarantine/avatars/")); + + const notYet = await read(bob, fileId); + assert.equal(notYet.status, 503); + assert.equal(notYet.headers.get("cache-control"), "no-store"); + + // The worker writes its copy and moves the row while a read is waiting. + const waiting = read(bob, fileId); + const fresh = Buffer.from("written out again by the worker"); + setTimeout(() => { + void putObject({ bucket: "gryt-test", key: `avatars/${fileId}.avif`, body: fresh, contentType: "image/avif" }).then(() => + getSqliteDb().prepare("UPDATE files SET s3_key = ?, mime = ?, size = ? WHERE file_id = ?").run(`avatars/${fileId}.avif`, "image/avif", fresh.length, fileId)); + }, 300); + const res = await waiting; + assert.equal(res.status, 200); + assert.deepEqual(Buffer.from(await res.arrayBuffer()), fresh); + }); + + it("keeps the old path when the worker does not say it clears quarantine", async () => { + assert.equal(await workerSays([]), false); + const carol = await member("carol"); + const fileId = (await upload(carol, "banner")).body.bannerFileId as string; + assert.ok((await getFile(fileId))?.s3_key.startsWith("banners/")); + }); +}); diff --git a/src/routes/uploads.ts b/src/routes/uploads.ts index 187f3f77..8e47b444 100644 --- a/src/routes/uploads.ts +++ b/src/routes/uploads.ts @@ -19,7 +19,12 @@ import { RangeNotSatisfiableError } from "../utils/byteRange"; import { sendStoredBody } from "../utils/sendStoredBody"; import { ensurePermission } from "../middleware/requirePermission"; import { sanitizeSvg } from "../utils/svgSanitize"; +import { quarantineUpload, waitOutOfQuarantine } from "../services/quarantineUpload"; + +/** How long a read waits for the worker to write a fresh upload out; a test sets it short. */ +const QUARANTINE_WAIT_MS = Number(process.env.GRYT_QUARANTINE_WAIT_MS) || 15_000; import { storeUploadedFile } from "../services/storeUploadedFile"; +import { workerClearsQuarantine } from "../services/workerCapabilities"; /** SVG is on this list only because it has been through sanitizeSvg(), is drawn through ``, and is sandboxed by the CSP below. That header is required. */ @@ -209,7 +214,17 @@ const storeAvatarImage = (purpose: "avatar" | "group" | "webhook" | "banner") => return; } - const stored = await storeAvatarPicture({ + // With a worker that clears quarantine, the server never decodes the file itself. + const stored = (purpose === "avatar" || purpose === "banner") && workerClearsQuarantine() + ? { ok: true as const, processing: true, ...(await quarantineUpload({ + bucket, + use: purpose === "avatar" ? "avatars" : "banners", + bytes: file.buffer, + mime: (file.mimetype || "application/octet-stream").toLowerCase(), + originalName: file.originalname || null, + uploadedBy: serverUserId, + })) } + : await storeAvatarPicture({ bucket, bytes: file.buffer, mime: file.mimetype || "", @@ -415,8 +430,15 @@ uploadsRouter.get( res.status(503).json({ error: "unavailable", message: "Could not check that just now. Try again in a moment." }); return; } - const fileMeta = verdict === "allowed" ? await getFile(fileId) : null; - if (!fileMeta) { res.status(404).json({ error: "File not found" }); return; } + const found = verdict === "allowed" ? await getFile(fileId) : null; + if (!found) { res.status(404).json({ error: "File not found" }); return; } + // Never the quarantined original: wait for the worker's copy, or say it isn't ready. + const fileMeta = await waitOutOfQuarantine(fileId, found, undefined, QUARANTINE_WAIT_MS); + if (!fileMeta) { + res.setHeader("Cache-Control", "no-store"); + res.status(503).json({ error: "media_not_ready", message: "This file is still being processed." }); + return; + } const useThumb = req.query.thumb === "1" && fileMeta.thumbnail_key; const s3Key = useThumb ? fileMeta.thumbnail_key! : fileMeta.s3_key; diff --git a/src/services/quarantineUpload.ts b/src/services/quarantineUpload.ts new file mode 100644 index 00000000..c6d28729 --- /dev/null +++ b/src/services/quarantineUpload.ts @@ -0,0 +1,58 @@ +import { v4 as uuidv4 } from "uuid"; + +import { getFile, insertFile, insertImageJob } from "../db"; +import { putObject } from "../storage"; + +export const QUARANTINE_PREFIX = "quarantine/"; + +export type QuarantineUse = "avatars" | "banners" | "uploads"; + +/** Stored as sent but out of reach: no route serves a quarantine key. The worker writes + it out again and moves the row, and the original is deleted then (GRYT-1664). */ +export async function quarantineUpload(input: { + bucket: string; + use: QuarantineUse; + bytes: Buffer; + mime: string; + originalName: string | null; + uploadedBy: string; +}): Promise<{ fileId: string; key: string }> { + const fileId = uuidv4(); + const key = `${QUARANTINE_PREFIX}${input.use}/${fileId}`; + await putObject({ bucket: input.bucket, key, body: input.bytes, contentType: input.mime }); + await insertFile({ + file_id: fileId, + s3_key: key, + mime: input.mime, + size: input.bytes.length, + width: null, + height: null, + thumbnail_key: null, + original_name: input.originalName, + uploaded_by_server_user_id: input.uploadedBy, + created_at: new Date(), + }); + await insertImageJob({ job_id: uuidv4(), file_id: fileId, raw_s3_key: key, raw_content_type: input.mime, raw_bytes: input.bytes.length }); + return { fileId, key }; +} + +export const isQuarantined = (key: string | null | undefined): boolean => !!key && key.startsWith(QUARANTINE_PREFIX); + +/** A file just uploaded is usually written out within a second or two, so a read waits + for it rather than failing. Null when it is still in quarantine after `waitMs`. */ +export async function waitOutOfQuarantine( + fileId: string, + first: T, + read: (id: string) => Promise = getFile as unknown as (id: string) => Promise, + waitMs = 15_000, + stepMs = 250, +): Promise { + let file: T | null = first; + const until = Date.now() + waitMs; + while (file && isQuarantined(file.s3_key)) { + if (Date.now() >= until) return null; + await new Promise((r) => setTimeout(r, stepMs)); + file = await read(fileId); + } + return file; +} diff --git a/src/services/storeUploadedFile.ts b/src/services/storeUploadedFile.ts index 45d8abc1..f7f188f7 100644 --- a/src/services/storeUploadedFile.ts +++ b/src/services/storeUploadedFile.ts @@ -5,6 +5,8 @@ import { v4 as uuidv4 } from "uuid"; import { insertFile, insertImageJob } from "../db"; import { storageForUpload } from "../routes/uploadStorage"; import { putObject } from "../storage"; +import { QUARANTINE_PREFIX } from "./quarantineUpload"; +import { workerClearsQuarantine } from "./workerCapabilities"; import { validateImage } from "../utils/imageValidation"; import { sanitizeSvg } from "../utils/svgSanitize"; import { PARSE_LIMITS, readVideoDimensionsFromFile } from "../utils/videoDimensions"; @@ -63,7 +65,10 @@ export async function storeUploadedFile(input: { }; } - const { key, storedMime } = storage; + const { storedMime } = storage; + // A plain image goes to quarantine when the worker clears it; videos keep today's path for now. + const quarantined = storage.queueImageJob && storage.validateAsImage && workerClearsQuarantine(); + const key = quarantined ? `${QUARANTINE_PREFIX}${storage.key}` : storage.key; let width: number | null = null; let height: number | null = null; @@ -94,7 +99,8 @@ export async function storeUploadedFile(input: { return { ok: true, fileId, key: svgKey }; } - if (storage.validateAsImage) { + // A quarantined image is decoded by the worker in its jail, never here. + if (storage.validateAsImage && !quarantined) { // The mime off the request is a claim, and taking it meant an SVG // carrying