From f11b7753f98b5db29d0682107e157f0e6e930f87 Mon Sep 17 00:00:00 2001 From: Sivert Date: Mon, 28 Sep 2026 10:29:11 +0200 Subject: [PATCH] Keep the files an MLS message carries (GRYT-1523) An MLS message never goes through chat:send, so no message row pointed at the files it carried, and the media sweep deleted them 30 minutes after upload. mls:send now takes attachmentIds. They're checked by chat:send's rules for a sealed DM: attach_files, at most ten, each one a file the sender can already read, and under the upload cap. Then they're recorded against the log entry in a new mls_attachments table. That table counts as a reference for the media sweep, and as somewhere a file was sent, so the other person can fetch it. Its rows go with the log entry, by a cascade. When retention drops an entry, its files are deleted straight away if nothing else holds them, the way a deleted message's are. Otherwise the media sweep takes them. Co-Authored-By: Claude Opus 5.5 --- src/db/sqlite/connection.ts | 11 +++++++ src/db/sqlite/messages.ts | 17 ++++++++--- src/db/sqlite/mls.test.ts | 18 ++++++++++++ src/db/sqlite/mls.ts | 19 +++++++++++- src/jobs/mlsRetention.ts | 8 ++++- src/socket/handlers/mls.test.ts | 45 +++++++++++++++++++++++++++- src/socket/handlers/mls.ts | 52 +++++++++++++++++++++++++++++---- 7 files changed, 158 insertions(+), 12 deletions(-) diff --git a/src/db/sqlite/connection.ts b/src/db/sqlite/connection.ts index d9e88a86..6a674565 100644 --- a/src/db/sqlite/connection.ts +++ b/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, diff --git a/src/db/sqlite/messages.ts b/src/db/sqlite/messages.ts index e01dba35..63df2712 100644 --- a/src/db/sqlite/messages.ts +++ b/src/db/sqlite/messages.ts @@ -204,8 +204,11 @@ export async function getAllReferencedAttachmentIds(): Promise> { 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; } @@ -213,7 +216,10 @@ export async function getAllReferencedAttachmentIds(): Promise> { /** Whether any message still points at a file, as an attachment or as something it shows. */ export async function isFileReferencedByMessage(fileId: string): Promise { 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 | undefined; diff --git a/src/db/sqlite/mls.test.ts b/src/db/sqlite/mls.test.ts index d9dd11dd..ea0adda2 100644 --- a/src/db/sqlite/mls.test.ts +++ b/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"); diff --git a/src/db/sqlite/mls.ts b/src/db/sqlite/mls.ts index e87d1be2..9c57a61f 100644 --- a/src/db/sqlite/mls.ts +++ b/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, }; }); } diff --git a/src/jobs/mlsRetention.ts b/src/jobs/mlsRetention.ts index 9f027e9a..e9d73b97 100644 --- a/src/jobs/mlsRetention.ts +++ b/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 { 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; } diff --git a/src/socket/handlers/mls.test.ts b/src/socket/handlers/mls.test.ts index bf101d55..6f3689be 100644 --- a/src/socket/handlers/mls.test.ts +++ b/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(); diff --git a/src/socket/handlers/mls.ts b/src/socket/handlers/mls.ts index 0a515423..dc4bdd65 100644 --- a/src/socket/handlers/mls.ts +++ b/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 { + async function droppedAsSpam(auth: AuthResult, conversationId: string, memberIds: string[], size: number, newConversation: boolean, attachments: number): Promise { 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 { + 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 }));