Keep the files an MLS message carries (GRYT-1523)

src/db/sqlite/connection.ts

@@ -630,6 +630,17 @@ function createSchema(d: DatabaseSync): void {
     );
     CREATE INDEX IF NOT EXISTS idx_mls_log_created ON mls_log(created_at);
 
+    -- Uploads an MLS message carries (GRYT-1523). The keys are inside the message; this
+    -- only keeps the media sweep off them, and they go with the log entry.
+    CREATE TABLE IF NOT EXISTS mls_attachments (
+      file_id TEXT NOT NULL,
+      group_id TEXT NOT NULL,
+      seq INTEGER NOT NULL,
+      PRIMARY KEY (file_id, group_id, seq),
+      FOREIGN KEY (group_id, seq) REFERENCES mls_log(group_id, seq) ON DELETE CASCADE
+    );
+    CREATE INDEX IF NOT EXISTS idx_mls_attachments_entry ON mls_attachments(group_id, seq);
+
     CREATE TABLE IF NOT EXISTS mls_welcomes (
       welcome_id TEXT PRIMARY KEY,
       server_user_id TEXT NOT NULL,

src/db/sqlite/messages.ts

@@ -204,16 +204,22 @@ export async function getAllReferencedAttachmentIds(): Promise<Set<string>> {
     const attachments: string[] = JSON.parse(row.attachments);
     for (const id of attachments) ids.add(id);
   }
-  // A card's pictures and a webhook's per-message avatar are only in the reference table.
-  const refs = db.prepare(`SELECT DISTINCT file_id FROM message_attachments`).all() as { file_id: string }[];
+  // A card's pictures and a webhook's per-message avatar are only in the reference table,
+  // and an MLS message's files only in its own (GRYT-1523).
+  const refs = db
+    .prepare(`SELECT file_id FROM message_attachments UNION SELECT file_id FROM mls_attachments`)
+    .all() as { file_id: string }[];
   for (const row of refs) ids.add(row.file_id);
   return ids;
 }
 
 /** Whether any message still points at a file, as an attachment or as something it shows. */
 export async function isFileReferencedByMessage(fileId: string): Promise<boolean> {
   const db = getSqliteDb();
-  return Boolean(db.prepare(`SELECT 1 FROM message_attachments WHERE file_id = ? LIMIT 1`).get(fileId));
+  return Boolean(
+    db.prepare(`SELECT 1 FROM message_attachments WHERE file_id = ? UNION ALL SELECT 1 FROM mls_attachments WHERE file_id = ? LIMIT 1`)
+      .get(fileId, fileId),
+  );
 }
 
 /** Everything that can make a file readable, in one indexed read. Null when
@@ -225,7 +231,10 @@ export async function getFileOwnership(fileId: string): Promise<FileOwnership |
        f.uploaded_by_server_user_id AS uploader,
        EXISTS (SELECT 1 FROM users u WHERE u.avatar_file_id = f.file_id) AS user_avatar,
        EXISTS (SELECT 1 FROM webhooks w WHERE w.avatar_file_id = f.file_id) AS webhook_avatar,
-       (SELECT json_group_array(DISTINCT a.conversation_id) FROM message_attachments a WHERE a.file_id = f.file_id) AS attached_to,
+       (SELECT json_group_array(DISTINCT t.conversation_id) FROM (
+          SELECT a.conversation_id FROM message_attachments a WHERE a.file_id = f.file_id
+          UNION SELECT g.conversation_id FROM mls_attachments m JOIN mls_groups g ON g.group_id = m.group_id
+           WHERE m.file_id = f.file_id) t) AS attached_to,
        (SELECT json_group_array(c.conversation_id) FROM conversations c WHERE c.icon_file_id = f.file_id) AS icon_of
      FROM files f WHERE f.file_id = ?`,
   ).get(fileId) as Record<string, unknown> | undefined;

src/db/sqlite/mls.test.ts

@@ -333,6 +333,24 @@ describe("retention", () => {
     assert.equal(getMlsGroup("aa99")?.epoch, 1, "sweeping the log leaves the epoch where it was");
   });
 
+  it("hands back the files a dropped entry held, and forgets them with it", async () => {
+    const conv = await conversation();
+    createMlsGroup("ae01", conv, "u1");
+    const base = { groupId: "ae01", senderServerUserId: "u1", senderDeviceId: "d" };
+    appendMlsMessage("application", { ...base, epoch: 0, data: bytes(7), attachmentIds: ["held-old"] }, new Date(Date.now() - 40 * DAY));
+    appendMlsMessage("application", { ...base, epoch: 0, data: bytes(8), attachmentIds: ["held-new"] });
+    const held = () => (getSqliteDb().prepare(`SELECT file_id FROM mls_attachments WHERE group_id = 'ae01'`).all() as { file_id: string }[]).map((r) => r.file_id);
+
+    const swept = sweepMls(new Date(Date.now() - 30 * DAY));
+    assert.ok(swept.fileIds.includes("held-old"));
+    assert.ok(!swept.fileIds.includes("held-new"));
+    assert.deepEqual(held(), ["held-new"]);
+
+    getSqliteDb().prepare(`DELETE FROM conversations WHERE conversation_id = ?`).run(conv);
+    assert.ok(sweepMls(new Date(Date.now() - 30 * DAY)).fileIds.includes("held-new"), "a group dropped with its conversation too");
+    assert.deepEqual(held(), []);
+  });
+
   it("drops a group whose conversation is gone, with its log", async () => {
     const conv = await conversation();
     createMlsGroup("ab01", conv, "u1");

src/db/sqlite/mls.ts

@@ -360,6 +360,8 @@ export interface MlsAppend {
   senderServerUserId: string;
   senderDeviceId: string;
   data: Uint8Array;
+  /** Upload ids the entry holds, checked by the caller. Kept until the entry goes. */
+  attachmentIds?: string[];
 }
 
 export interface MlsWelcomeFor {
@@ -453,6 +455,8 @@ export function appendMlsMessage(
     const seq = group.headSeq + 1;
     insertEntry(kind, a, seq, at);
     getSqliteDb().prepare(`UPDATE mls_groups SET head_seq = ? WHERE group_id = ?`).run(seq, a.groupId);
+    const ref = getSqliteDb().prepare(`INSERT OR IGNORE INTO mls_attachments (file_id, group_id, seq) VALUES (?, ?, ?)`);
+    for (const fileId of a.attachmentIds ?? []) ref.run(fileId, a.groupId, seq);
     return { accepted: true as const, seq, createdAt: at, duplicate: false };
   });
 }
@@ -512,10 +516,22 @@ export function deleteMlsWelcomes(serverUserId: string, deviceId: string, welcom
 export function sweepMls(
   cutoff: Date,
   now = new Date(),
-): { log: number; welcomes: number; keyPackages: number; expired: number; groups: number } {
+): { log: number; welcomes: number; keyPackages: number; expired: number; groups: number; fileIds: string[] } {
   return inTransaction(() => {
     const db = getSqliteDb();
     const before = toIso(cutoff);
+    // What the dropped entries held, for the caller to delete once nothing else holds it.
+    const fileIds = (
+      db
+        .prepare(
+          `SELECT DISTINCT a.file_id FROM mls_attachments a
+             JOIN mls_log l ON l.group_id = a.group_id AND l.seq = a.seq
+             LEFT JOIN mls_groups g ON g.group_id = l.group_id
+             LEFT JOIN conversations c ON c.conversation_id = g.conversation_id
+            WHERE l.created_at < ? OR c.conversation_id IS NULL`,
+        )
+        .all(before) as { file_id: string }[]
+    ).map((r) => r.file_id);
     // Expired ones lose their bytes like a replaced last-resort one, so a Welcome still routes.
     const expired = db
       .prepare(
@@ -542,6 +558,7 @@ export function sweepMls(
       keyPackages: Number(keyPackages.changes),
       expired: Number(expired.changes),
       groups: orphans.length,
+      fileIds,
     };
   });
 }

src/jobs/mlsRetention.ts

@@ -1,6 +1,7 @@
 import consola from "consola";
 
 import { sweepMls } from "../db";
+import { deleteUnreferencedFiles } from "./mediaSweep";
 
 /** Decision 7 in docs/mls-design.md: 30 days, and a host can make it shorter but not longer. */
 export const MLS_MAX_RETENTION_DAYS = 30;
@@ -20,8 +21,13 @@ export function mlsRetentionDays(env: NodeJS.ProcessEnv = process.env): number {
 
 export function runMlsRetention(now = new Date()): ReturnType<typeof sweepMls> {
   const swept = sweepMls(new Date(now.getTime() - mlsRetentionDays() * DAY_MS), now);
+  const { fileIds, ...counts } = swept;
   if (swept.log || swept.welcomes || swept.keyPackages || swept.expired || swept.groups) {
-    consola.info("[mls-retention] swept", swept);
+    consola.info("[mls-retention] swept", { ...counts, files: fileIds.length });
+  }
+  // As a deleted message's files go: now if nothing else holds them, else the media sweep's.
+  if (fileIds.length > 0) {
+    void deleteUnreferencedFiles(fileIds).catch((err) => consola.warn("[mls-retention] file cleanup failed", err));
   }
   return swept;
 }

src/socket/handlers/mls.test.ts

@@ -15,12 +15,15 @@ import { createGroupConversation, openDirectConversation } from "../../db/sqlite
 import { getMlsGroupForConversation } from "../../db/sqlite/mls";
 import { createServerConfigIfNotExists, setServerRole, updateServerConfig } from "../../db/sqlite/servers";
 import { hasWrittenTo } from "../../db/sqlite/contactPrefs";
-import { listMessages } from "../../db/sqlite/messages";
+import { insertFile, listMessages } from "../../db/sqlite/messages";
 import { setUserModerationState, upsertUser } from "../../db/sqlite/users";
+import { unreferencedAmong } from "../../jobs/mediaSweep";
 import { mlsRetentionDays, runMlsRetention } from "../../jobs/mlsRetention";
+import { fileReadVerdict, resetFileAccessCache } from "../../services/fileAccess";
 import { checkKeyPackageLifetime, parseGroupMessage, parseKeyPackage, parseWelcome } from "../../services/mlsWire";
 import type { Clients } from "../../types";
 import { generateAccessToken } from "../../utils/jwt";
+import { resetRateLimits } from "../../utils/rateLimiter";
 import { resetChannelIdCache } from "../utils/conversationAccess";
 import { refreshClientPermissions } from "../utils/standing";
 import { registerChatHandlers } from "./chat";
@@ -707,6 +710,34 @@ describe("a DM over MLS", () => {
     const odd = await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: await applicationMessage(a1, "x"), placeholder: "no" });
     assert.equal(odd.error, "invalid_payload");
   });
+
+  it("carries files, and keeps them from the media sweep while the entry is there", async () => {
+    const upload = (id: string, by: string) => insertFile({
+      file_id: id, s3_key: `uploads/${id}`, mime: "application/octet-stream", size: 10, width: null, height: null,
+      thumbnail_key: null, original_name: `${id}.bin`, uploaded_by_server_user_id: by,
+    });
+    await upload("mls-file-alice", alice.serverUserId);
+    await upload("mls-file-bob", bob.serverUserId);
+    resetFileAccessCache();
+    resetRateLimits();
+    assert.equal(await fileReadVerdict("mls-file-alice", bob.serverUserId), "denied");
+    assert.deepEqual(await unreferencedAmong(["mls-file-alice"]), ["mls-file-alice"]);
+
+    const send = async (attachmentIds: unknown, placeholder = false) =>
+      alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: await applicationMessage(a1, "a file"), placeholder, attachmentIds });
+    const sent = await send(["mls-file-alice", "mls-file-alice"]);
+    assert.equal(sent.ok, true, "a message with no line for older apps can carry files too");
+
+    assert.deepEqual(await unreferencedAmong(["mls-file-alice"]), [], "the media sweep leaves it alone");
+    assert.equal(await fileReadVerdict("mls-file-alice", bob.serverUserId), "allowed", "and the other side can fetch it");
+    const refs = getSqliteDb().prepare(`SELECT seq FROM mls_attachments WHERE file_id = ?`).all("mls-file-alice") as { seq: number }[];
+    assert.deepEqual(refs.map((r) => r.seq), [sent.seq]);
+
+    assert.equal((await send(["mls-file-bob"])).error, "attachment_not_found", "only files the sender can read");
+    assert.equal((await send(["no-such-upload"])).error, "attachment_not_found");
+    assert.equal((await send(Array.from({ length: 11 }, (_, i) => `f${i}`))).error, "too_many_attachments");
+    assert.equal((await send("mls-file-alice")).error, "invalid_payload");
+  });
 });
 
 describe("retention", () => {
@@ -718,6 +749,18 @@ describe("retention", () => {
     assert.equal(mlsRetentionDays({ MLS_RETENTION_DAYS: "soon" }), 30);
   });
 
+  it("lets an entry's files go when the entry is swept", async () => {
+    const group = getMlsGroupForConversation(dm)!;
+    const { seq } = getSqliteDb().prepare(`SELECT seq FROM mls_attachments WHERE file_id = ?`).get("mls-file-alice") as { seq: number };
+    const old = new Date(Date.now() - 40 * 24 * 60 * 60 * 1000).toISOString();
+    getSqliteDb().prepare(`UPDATE mls_log SET created_at = ? WHERE group_id = ? AND seq = ?`).run(old, group.groupId, seq);
+
+    assert.ok(runMlsRetention().fileIds.includes("mls-file-alice"));
+    assert.deepEqual(await unreferencedAmong(["mls-file-alice"]), ["mls-file-alice"], "the media sweep can have it now");
+    resetFileAccessCache();
+    assert.equal(await fileReadVerdict("mls-file-alice", bob.serverUserId), "denied");
+  });
+
   it("sweeps old ciphertext, and a device behind it is told it has a gap", async () => {
     const group = getMlsGroupForConversation(dm)!;
     const old = new Date(Date.now() - 10 * 24 * 60 * 60 * 1000).toISOString();

src/socket/handlers/mls.ts

@@ -6,12 +6,15 @@ import {
   appendMlsMessage,
   blockedServerIdsFor,
   blockersOfSender,
+  DEFAULT_MAX_ATTACHMENTS_PER_MESSAGE as MAX_ATTACHMENTS_PER_MESSAGE,
+  DEFAULT_UPLOAD_MAX_BYTES,
   claimMlsKeyPackage,
   countMlsKeyPackages,
   createMlsGroup,
   deleteMlsWelcomes,
   eitherHasBlocked,
   getConversation,
+  getFilesByIds,
   getMlsGroupForConversation,
   getServerConfig,
   getUserByServerId,
@@ -36,6 +39,7 @@ import { spamFilter } from "../../moderation/spamFilter";
 import { isSpamExempt, spamRefusal, timeOutSpammer } from "../../moderation/spamTimeout";
 import { textMuteError, textMuteFor } from "../../moderation/textMute";
 import { mayInChannel } from "../../services/channelPermissions";
+import { fileReadVerdict } from "../../services/fileAccess";
 import { asBytes, parseGroupMessage, parseKeyPackage, parseWelcome } from "../../services/mlsWire";
 import { SEALED_MAX_LENGTH } from "../../utils/messageLimits";
 import { appendCachedMessage } from "../utils/messageCache";
@@ -191,7 +195,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
   }
 
   /* Metadata only, as for sealed DMs: who, how often, how big. */
-  async function droppedAsSpam(auth: AuthResult, conversationId: string, memberIds: string[], size: number, newConversation: boolean): Promise<Reply | null> {
+  async function droppedAsSpam(auth: AuthResult, conversationId: string, memberIds: string[], size: number, newConversation: boolean, attachments: number): Promise<Reply | null> {
     const cfg = await getServerConfig().catch(() => null);
     if (cfg && cfg.spam_filter_enabled === false) return null;
     if (isSpamExempt({ isOwner: auth.isOwner, permissions: auth.permissions, grytUserId: auth.tokenPayload.grytUserId })) return null;
@@ -207,7 +211,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
         recipients: memberIds.filter((id) => id !== user.server_user_id),
         size,
         newConversation,
-        attachments: 0,
+        attachments,
       },
       sensitivity,
     );
@@ -218,6 +222,31 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
     return { ok: false, ...spamRefusal(until) };
   }
 
+  /* chat:send's rules for a sealed DM's files (GRYT-1523): the permission, the count, and
+     each one a file the sender can already read and under the size cap. Null is yes. */
+  async function attachmentRefusal(auth: AuthResult, conversationId: string, ids: string[]): Promise<Reply | null> {
+    if (ids.length === 0) return null;
+    if (ids.length > MAX_ATTACHMENTS_PER_MESSAGE) {
+      return fail("too_many_attachments", `A message can carry at most ${MAX_ATTACHMENTS_PER_MESSAGE} files.`);
+    }
+    if (!(await mayInChannel(conversationId, auth.tokenPayload.serverUserId, "attach_files", auth.tokenPayload.grytUserId))) {
+      return fail("forbidden", "You do not have permission to attach files here.", { permission: "attach_files" });
+    }
+    const cfg = await getServerConfig().catch(() => null);
+    const maxBytes = typeof cfg?.upload_max_bytes === "number" ? cfg.upload_max_bytes : DEFAULT_UPLOAD_MAX_BYTES;
+    const files = await getFilesByIds(ids);
+    for (const id of ids) {
+      const f = files.get(id);
+      if (!f || (await fileReadVerdict(id, auth.tokenPayload.serverUserId, auth.tokenPayload.grytUserId)) !== "allowed") {
+        return fail("attachment_not_found", `Attachment not found: ${id}`);
+      }
+      if (typeof maxBytes === "number" && maxBytes > 0 && f.size != null && f.size > maxBytes) {
+        return fail("attachment_too_large", `File "${f.original_name || id}" is too large. Max ${(maxBytes / (1024 * 1024)).toFixed(1)}MB.`);
+      }
+    }
+    return null;
+  }
+
   /* Every device of a member, or refused: a group can only grow by its own people. */
   function memberDevicesFor(refs: string[], memberIds: string[]): { serverUserId: string; deviceId: string }[] | null {
     const members = new Set(memberIds);
@@ -557,7 +586,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
 
     /** An application message (PrivateMessage) or a proposal (PublicMessage). */
     "mls:send": async (
-      payload: { accessToken: string; conversationId: string; deviceId: string; message: unknown; placeholder?: unknown },
+      payload: { accessToken: string; conversationId: string; deviceId: string; message: unknown; placeholder?: unknown; attachmentIds?: unknown },
       ack: Ack,
     ) => {
       ack = typeof ack === "function" ? ack : () => {};
@@ -568,6 +597,12 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
           ack(fail("invalid_payload", "placeholder has to be true or false."));
           return;
         }
+        const raw = payload.attachmentIds;
+        if (raw !== undefined && (!Array.isArray(raw) || !raw.every((id) => typeof id === "string" && id.length > 0 && id.length <= 128))) {
+          ack(fail("invalid_payload", "attachmentIds has to be a list of upload ids."));
+          return;
+        }
+        const attachmentIds = [...new Set((raw ?? []) as string[])];
         const self = auth.tokenPayload.serverUserId;
         const memberIds = await dmOf(payload.conversationId, self, ack);
         if (!memberIds) return;
@@ -601,15 +636,21 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
         }
 
         const application = parsed.kind === "application";
+        if (!application && attachmentIds.length > 0) {
+          ack(fail("invalid_payload", "Only an application message can carry files."));
+          return;
+        }
         let wasEmpty = false;
         if (application) {
-          const refusal = await sendRefusal(auth, payload.conversationId, memberIds);
+          const refusal =
+            (await sendRefusal(auth, payload.conversationId, memberIds)) ??
+            (await attachmentRefusal(auth, payload.conversationId, attachmentIds));
           if (refusal) {
             ack(refusal);
             return;
           }
           wasEmpty = !(await getConversation(payload.conversationId))?.last_message_at;
-          const spam = await droppedAsSpam(auth, payload.conversationId, memberIds, bytes.length, wasEmpty);
+          const spam = await droppedAsSpam(auth, payload.conversationId, memberIds, bytes.length, wasEmpty, attachmentIds.length);
           if (spam) {
             ack(spam);
             return;
@@ -618,6 +659,7 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap {
 
         const result = appendMlsMessage(parsed.kind, {
           groupId: group.groupId, epoch: parsed.epoch, senderServerUserId: self, senderDeviceId: payload.deviceId, data: bytes,
+          attachmentIds,
         });
         if (!result.accepted) {
           if (result.reason === "future_epoch") ack(fail("future_epoch", "That epoch hasn't happened yet.", { epoch: result.epoch }));