Avatars, banners and chat images through the worker's quarantine (GRYT-1664)

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)

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<Member> {
+  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<string, unknown> };
+}
+
+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<void>((resolve) => server.once("listening", () => resolve()));
+  host = `127.0.0.1:${(server.address() as AddressInfo).port}`;
+  base = `http://${host}`;
+});
+
+after(async () => {
+  await new Promise<void>((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/"));
+  });
+});

src/routes/uploadStorage.test.ts

@@ -172,7 +172,8 @@ describe("the route uses it", () => {
     // sharp, the worker and the SVG sanitiser would each be handed it.
     for (const [what, gate] of [
       ["the SVG sanitiser", "if (storage.treatAsSvg)"],
-      ["image validation", "if (storage.validateAsImage)"],
+      // Skipped too for a quarantined image, which the worker decodes in its jail (GRYT-1664).
+      ["image validation", "if (storage.validateAsImage && !quarantined)"],
       ["the video container parser", "if (storage.measureAsVideo)"],
       ["the image worker", "if (storage.queueImageJob)"],
     ]) {

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 `<img>`, 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;

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<T extends { s3_key: string }>(
+  fileId: string,
+  first: T,
+  read: (id: string) => Promise<T | null> = getFile as unknown as (id: string) => Promise<T | null>,
+  waitMs = 15_000,
+  stepMs = 250,
+): Promise<T | null> {
+  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;
+}

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 <script> was served back inline.
     if (size > IMAGE_VALIDATION_MAX_BYTES) {

src/services/workerCapabilities.ts

@@ -0,0 +1,32 @@
+import consola from "consola";
+
+/* Whether the image worker can clear quarantine: it says so on /health once it has its jail
+   (GRYT-1664). Asked every minute; any doubt reads as no, which keeps the old path. */
+const REFRESH_MS = 60_000;
+
+let clears = false;
+let timer: NodeJS.Timeout | null = null;
+
+export function workerClearsQuarantine(): boolean {
+  return clears;
+}
+
+export async function refreshWorkerCapabilities(url = process.env.IMAGE_WORKER_URL, fetchImpl: typeof fetch = fetch): Promise<boolean> {
+  if (!url) return (clears = false);
+  try {
+    const res = await fetchImpl(`${url.replace(/\/$/, "")}/health`, { signal: AbortSignal.timeout(5_000) });
+    const body = res.ok ? ((await res.json()) as { capabilities?: unknown }) : null;
+    const next = Array.isArray(body?.capabilities) && body.capabilities.includes("quarantine-v1");
+    if (next !== clears) consola.info(`[uploads] Quarantine through the image worker ${next ? "on" : "off"}`);
+    return (clears = next);
+  } catch {
+    return (clears = false);
+  }
+}
+
+export function startWorkerCapabilityPolling(): void {
+  if (timer) return;
+  void refreshWorkerCapabilities();
+  timer = setInterval(() => void refreshWorkerCapabilities(), REFRESH_MS);
+  timer.unref();
+}