From 7504c7fb6120fb3404e97077f46463cdd096da24 Mon Sep 17 00:00:00 2001 From: Sivert Date: Fri, 25 Sep 2026 12:08:58 +0200 Subject: [PATCH 1/2] The MLS delivery service over the socket (GRYT-1500) Stage 1's server half from docs/mls-design.md: KeyPackages, Welcomes, a log per DM group with cursors, one commit per epoch, and a flag saying the server has all of this. Every request answers through its socket.io ack. The server reads a few header fields and nothing else: group id, epoch and content type, the refs of the KeyPackages a commit adds, and who a Welcome is for. It gets them from ts-mls 1.6.4, pinned exactly. It loads only the message and keyPackage subpaths, so the noble provider is never pulled in. It uses those fields to: - refuse a commit for any epoch but the current one, with the current one - refuse a message from an epoch that hasn't happened yet - refuse an Add or a Welcome for a device that doesn't belong to somebody in the conversation - hand each Welcome to the device it names, without the client saying who Commits and proposals have to be PublicMessage, or the Adds couldn't be checked. Application messages have to be PrivateMessage. Only one-to-one DMs for now. Group DMs are stage 2 and bring their own checks. Everything a sealed DM goes through still applies to an MLS message: mute, send_messages, the server's DM switch, send_direct_messages, contact settings, and the spam filter on size and recipients alone. Claiming somebody's KeyPackages goes through the same checks. A block either way reads as "not a member", as it does for dm:open. Somebody who blocked the sender doesn't get their messages, live or from the log. They still get the sender's commits, or their group state would fall behind. server:info carries mls: { version: 1, ciphersuites: [1], retentionDays }. The log is kept for 30 days, or fewer if MLS_RETENTION_DAYS says so, and swept hourly. A device whose cursor is older than what's left is told it has a gap. Co-Authored-By: Claude Opus 5.5 --- example.env | 5 + package.json | 1 + src/index.ts | 2 + src/jobs/mlsRetention.ts | 41 ++ src/services/mlsWire.ts | 134 ++++++ src/socket/handlers/mls.test.ts | 570 ++++++++++++++++++++++ src/socket/handlers/mls.ts | 709 ++++++++++++++++++++++++++++ src/socket/index.ts | 2 + src/socket/utils/server.ts | 5 + src/socket/utils/serverInfo.test.ts | 10 + yarn.lock | 19 + 11 files changed, 1498 insertions(+) create mode 100644 src/jobs/mlsRetention.ts create mode 100644 src/services/mlsWire.ts create mode 100644 src/socket/handlers/mls.test.ts create mode 100644 src/socket/handlers/mls.ts diff --git a/example.env b/example.env index cb6cc919..2e2014b0 100644 --- a/example.env +++ b/example.env @@ -92,6 +92,11 @@ JWT_SECRET=change-me-in-production # Database (SQLite, stored in DATA_DIR) DATA_DIR=./data +# How long the server keeps encrypted (MLS) direct messages, which it can't read. +# 30 days at most, and 30 if unset. A device that's offline for longer misses what +# was sent in the meantime. +# MLS_RETENTION_DAYS=30 + # Unique instance ID (used in multi-instance setups to distinguish servers) # SERVER_INSTANCE_ID=default diff --git a/package.json b/package.json index ef39a204..3dc551fb 100644 --- a/package.json +++ b/package.json @@ -25,6 +25,7 @@ "prom-client": "^15.1.3", "sharp": "^0.35.0", "socket.io": "^4.8.3", + "ts-mls": "1.6.4", "undici": "^8.9.0", "uuid": "^11.1.0", "ws": "^8.19.0", diff --git a/src/index.ts b/src/index.ts index 8d4bb343..8f552af5 100644 --- a/src/index.ts +++ b/src/index.ts @@ -39,6 +39,7 @@ import { parsesOwnJson } from "./routes/ownJsonParsers"; import { apiErrorHandler, jsonBodyExcept } from "./utils/httpErrors"; import { sendStoredBody } from "./utils/sendStoredBody"; import { startMediaSweep } from "./jobs/mediaSweep"; +import { startMlsRetention } from "./jobs/mlsRetention"; import { startEmojiQueueWorker } from "./jobs/emojiQueueWorker"; import { initPlugins } from "./plugins"; import { @@ -134,6 +135,7 @@ initSqlite() }) .then(() => { if (!disableS3) startMediaSweep(); + startMlsRetention(); if (!disableS3 && (process.env.S3_BUCKET || "").trim()) { startEmojiQueueWorker(); } diff --git a/src/jobs/mlsRetention.ts b/src/jobs/mlsRetention.ts new file mode 100644 index 00000000..e18a4459 --- /dev/null +++ b/src/jobs/mlsRetention.ts @@ -0,0 +1,41 @@ +import consola from "consola"; + +import { sweepMls } from "../db"; + +/** 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; + +const DAY_MS = 24 * 60 * 60 * 1000; +const SWEEP_INTERVAL_MS = 60 * 60 * 1000; + +let timer: ReturnType | null = null; + +/** `MLS_RETENTION_DAYS`, whole days from 1 to 30. Anything else is read as 30. */ +export function mlsRetentionDays(env: NodeJS.ProcessEnv = process.env): number { + const raw = env.MLS_RETENTION_DAYS?.trim(); + if (!raw || !/^\d+$/.test(raw)) return MLS_MAX_RETENTION_DAYS; + const days = Number(raw); + return days >= 1 && days <= MLS_MAX_RETENTION_DAYS ? days : MLS_MAX_RETENTION_DAYS; +} + +export function runMlsRetention(now = new Date()): ReturnType { + const swept = sweepMls(new Date(now.getTime() - mlsRetentionDays() * DAY_MS)); + if (swept.log || swept.welcomes || swept.keyPackages || swept.groups) { + consola.info("[mls-retention] swept", swept); + } + return swept; +} + +export function startMlsRetention(): void { + if (timer) return; + const sweep = () => { + try { + runMlsRetention(); + } catch (err) { + consola.error("[mls-retention] sweep failed", err); + } + }; + sweep(); + timer = setInterval(sweep, SWEEP_INTERVAL_MS); + timer.unref?.(); +} diff --git a/src/services/mlsWire.ts b/src/services/mlsWire.ts new file mode 100644 index 00000000..ec7920c3 --- /dev/null +++ b/src/services/mlsWire.ts @@ -0,0 +1,134 @@ +import { createHash, createHmac, timingSafeEqual } from "crypto"; + +import type { KeyPackage } from "ts-mls/dist/src/keyPackage"; +import type { MLSMessage } from "ts-mls/dist/src/message"; + +/** + * The few MLS header fields the delivery service reads (GRYT-1500). Decoding only: + * nothing here holds a key, checks a signature or opens a ciphertext. + */ + +/* Through the package's exports map, which only these subpaths get past. The index + would pull in the noble provider the server has no use for. */ + +/* eslint-disable @typescript-eslint/no-require-imports */ +const { decodeMlsMessage } = require("ts-mls/message.js") as typeof import("ts-mls/dist/src/message"); +const { makeKeyPackageRef } = require("ts-mls/keyPackage.js") as typeof import("ts-mls/dist/src/keyPackage"); +/* eslint-enable @typescript-eslint/no-require-imports */ + +/** Suite 1, the only one Gryt uses (decision 1 in docs/mls-design.md). */ +export const MLS_CIPHERSUITE = "MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519"; +export const MLS_CIPHERSUITE_ID = 1; + +export type WireRefusal = { error: string; message: string }; + +const refuse = (error: string, message: string): { ok: false } & WireRefusal => ({ ok: false, error, message }); + +/** Socket.io hands binary over as a Buffer, and some clients send an ArrayBuffer. */ +export function asBytes(value: unknown): Uint8Array | null { + if (value instanceof Uint8Array) return value; + if (value instanceof ArrayBuffer) return new Uint8Array(value); + return null; +} + +/* The whole input or nothing: trailing bytes would be stored and fanned out + without anybody having looked at them. */ +function decodeWhole(bytes: Uint8Array): MLSMessage | null { + try { + const decoded = decodeMlsMessage(bytes, 0); + if (!decoded || decoded[1] !== bytes.length || decoded[0].version !== "mls10") return null; + return decoded[0]; + } catch { + return null; + } +} + +const sha256: Parameters[1] = { + digest: async (data) => new Uint8Array(createHash("sha256").update(data).digest()), + mac: async (key, data) => new Uint8Array(createHmac("sha256", key).update(data).digest()), + verifyMac: async (key, mac, data) => { + const want = createHmac("sha256", key).update(data).digest(); + return want.length === mac.length && timingSafeEqual(want, mac); + }, +}; + +export async function keyPackageRef(kp: KeyPackage): Promise { + return Buffer.from(await makeKeyPackageRef(kp, sha256)).toString("hex"); +} + +export async function parseKeyPackage( + bytes: Uint8Array, +): Promise<{ ok: true; ref: string } | ({ ok: false } & WireRefusal)> { + const msg = decodeWhole(bytes); + if (!msg || msg.wireformat !== "mls_key_package") return refuse("invalid_key_package", "That isn't an MLS KeyPackage."); + if (msg.keyPackage.cipherSuite !== MLS_CIPHERSUITE) { + return refuse("unsupported_ciphersuite", `Only ${MLS_CIPHERSUITE} is accepted here.`); + } + return { ok: true, ref: await keyPackageRef(msg.keyPackage) }; +} + +export interface ParsedHandshake { + kind: "commit" | "proposal" | "application"; + groupId: string; + epoch: number; + /** Refs of the KeyPackages a commit or proposal adds by value. */ + addedRefs: string[]; + /** Proposals a commit names by reference; their Adds were checked on arrival. */ + referencedProposals: number; +} + +/** + * Commits and proposals as PublicMessage, so their Adds can be checked; application + * messages as PrivateMessage, so nothing about them is readable (docs/mls-design.md ยง2). + */ +export async function parseGroupMessage( + bytes: Uint8Array, +): Promise<({ ok: true } & ParsedHandshake) | ({ ok: false } & WireRefusal)> { + const msg = decodeWhole(bytes); + if (!msg) return refuse("invalid_message", "That isn't an MLS message."); + + if (msg.wireformat === "mls_private_message") { + const pm = msg.privateMessage; + if (pm.contentType !== "application") { + return refuse("must_be_public", "Commits and proposals have to be sent as PublicMessage."); + } + return { ok: true, kind: "application", groupId: hex(pm.groupId), epoch: Number(pm.epoch), addedRefs: [], referencedProposals: 0 }; + } + + if (msg.wireformat !== "mls_public_message") return refuse("invalid_message", "That isn't a group message."); + const content = msg.publicMessage.content; + if (content.sender.senderType !== "member") { + return refuse("unsupported_sender", "Only a member of the group can send a commit or proposal here."); + } + const base = { groupId: hex(content.groupId), epoch: Number(content.epoch) }; + + if (content.contentType === "proposal") { + const p = content.proposal; + const addedRefs = p.proposalType === "add" ? [await keyPackageRef(p.add.keyPackage)] : []; + return { ok: true, kind: "proposal", ...base, addedRefs, referencedProposals: 0 }; + } + if (content.contentType === "commit") { + const addedRefs: string[] = []; + let referencedProposals = 0; + for (const entry of content.commit.proposals) { + if (entry.proposalOrRefType === "reference") referencedProposals += 1; + else if (entry.proposal.proposalType === "add") addedRefs.push(await keyPackageRef(entry.proposal.add.keyPackage)); + } + return { ok: true, kind: "commit", ...base, addedRefs, referencedProposals }; + } + return refuse("must_be_private", "Application messages have to be sent as PrivateMessage."); +} + +/** Who a Welcome is for, as KeyPackage refs. */ +export function parseWelcome(bytes: Uint8Array): { ok: true; recipients: string[] } | ({ ok: false } & WireRefusal) { + const msg = decodeWhole(bytes); + if (!msg || msg.wireformat !== "mls_welcome") return refuse("invalid_welcome", "That isn't an MLS Welcome."); + if (msg.welcome.cipherSuite !== MLS_CIPHERSUITE) { + return refuse("unsupported_ciphersuite", `Only ${MLS_CIPHERSUITE} is accepted here.`); + } + return { ok: true, recipients: msg.welcome.secrets.map((s) => hex(s.newMember)) }; +} + +export function hex(bytes: Uint8Array): string { + return Buffer.from(bytes).toString("hex"); +} diff --git a/src/socket/handlers/mls.test.ts b/src/socket/handlers/mls.test.ts new file mode 100644 index 00000000..340c2fdf --- /dev/null +++ b/src/socket/handlers/mls.test.ts @@ -0,0 +1,570 @@ +import assert from "node:assert/strict"; +import { randomBytes } from "node:crypto"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { after, before, describe, it } from "node:test"; + +import type { ClientState } from "ts-mls/dist/src/clientState"; +import type { KeyPackage, PrivateKeyPackage } from "ts-mls/dist/src/keyPackage"; + +import { blockUser, unblockUser } from "../../db/sqlite/blocks"; +import { getSqliteDb, initSqlite } from "../../db/sqlite/connection"; +import { setContactPrefs } from "../../db/sqlite/contactPrefs"; +import { createGroupConversation, openDirectConversation } from "../../db/sqlite/conversations"; +import { getMlsGroupForConversation } from "../../db/sqlite/mls"; +import { createServerConfigIfNotExists, setServerRole, updateServerConfig } from "../../db/sqlite/servers"; +import { setUserModerationState, upsertUser } from "../../db/sqlite/users"; +import { mlsRetentionDays, runMlsRetention } from "../../jobs/mlsRetention"; +import { parseGroupMessage, parseKeyPackage, parseWelcome } from "../../services/mlsWire"; +import type { Clients } from "../../types"; +import { generateAccessToken } from "../../utils/jwt"; +import { resetChannelIdCache } from "../utils/conversationAccess"; +import { refreshClientPermissions } from "../utils/standing"; +import { registerMlsHandlers } from "./mls"; +import type { EventHandlerMap, HandlerContext } from "./types"; + +/** Real suite 1 MLS from ts-mls on both ends, so what the server stores and hands + back is proven to still open, not just to be the same length. */ + +/* eslint-disable @typescript-eslint/no-require-imports */ +const mls = { + ...(require("ts-mls/clientState.js") as typeof import("ts-mls/dist/src/clientState")), + ...(require("ts-mls/createCommit.js") as typeof import("ts-mls/dist/src/createCommit")), + ...(require("ts-mls/createMessage.js") as typeof import("ts-mls/dist/src/createMessage")), + ...(require("ts-mls/processMessages.js") as typeof import("ts-mls/dist/src/processMessages")), + ...(require("ts-mls/keyPackage.js") as typeof import("ts-mls/dist/src/keyPackage")), + ...(require("ts-mls/message.js") as typeof import("ts-mls/dist/src/message")), + ...(require("ts-mls/crypto/getCiphersuiteImpl.js") as typeof import("ts-mls/dist/src/crypto/getCiphersuiteImpl")), + ...(require("ts-mls/crypto/ciphersuite.js") as typeof import("ts-mls/dist/src/crypto/ciphersuite")), + ...(require("ts-mls/defaultCapabilities.js") as typeof import("ts-mls/dist/src/defaultCapabilities")), + ...(require("ts-mls/lifetime.js") as typeof import("ts-mls/dist/src/lifetime")), + ...(require("ts-mls/pskIndex.js") as typeof import("ts-mls/dist/src/pskIndex")), +}; +/* eslint-enable @typescript-eslint/no-require-imports */ + +type Impl = Awaited>; +let impl: Impl; + +const HOST = "mls.test:5001"; +let dir: string; + +interface Reply { + ok: boolean; + error?: string; + [key: string]: unknown; +} + +interface Participant { + serverUserId: string; + grytUserId: string; + accessToken: string; + handlers: EventHandlerMap; + received: (event: string) => Record[]; + clear: () => void; + call: (event: string, payload?: Record) => Promise; +} + +const clientsInfo: Clients = {}; +const sockets = new Map boolean }>(); +const io = { to: () => ({ emit() {} }), emit() {}, sockets: { sockets } }; + +let seq = 0; +async function connect(nickname: string): Promise { + seq += 1; + const clientId = `mls-socket-${seq}`; + const grytUserId = `account-mls-${seq}`; + const user = await upsertUser(grytUserId, nickname); + await setServerRole(user.server_user_id, "member"); + + const emitted: { event: string; payload: unknown }[] = []; + const emit = (event: string, payload?: unknown) => { + emitted.push({ event, payload }); + return true; + }; + sockets.set(clientId, { emit }); + clientsInfo[clientId] = { + serverUserId: user.server_user_id, grytUserId, nickname, color: "#666666", + isMuted: false, isDeafened: false, streamID: "", hasJoinedChannel: false, voiceChannelId: "", isAFK: false, + cameraEnabled: false, cameraStreamID: "", screenShareEnabled: false, screenShareVideoStreamID: "", + screenShareAudioStreamID: "", isServerMuted: false, isServerDeafened: false, + } as Clients[string]; + await refreshClientPermissions(clientsInfo, clientId); + + const ctx = { + io, + socket: { id: clientId, handshake: { headers: { host: HOST }, address: "127.0.0.1" }, emit, join() {}, leave() {} }, + clientId, + serverId: "mls-test", + clientsInfo, + sfuClient: null, + getClientIp: () => `10.9.0.${seq}`, + clientAddressIsOwn: () => true, + } as unknown as HandlerContext; + const handlers = registerMlsHandlers(ctx); + const accessToken = generateAccessToken({ grytUserId, serverUserId: user.server_user_id, nickname, serverHost: HOST, tokenVersion: 0 }); + + return { + serverUserId: user.server_user_id, + grytUserId, + accessToken, + handlers, + received: (event) => emitted.filter((e) => e.event === event).map((e) => e.payload as Record), + clear: () => { + emitted.length = 0; + }, + call: (event, payload = {}) => + new Promise((resolve) => { + void handlers[event]({ accessToken, ...payload }, resolve); + }), + }; +} + +/** One device's MLS side, as a client would hold it. */ +interface Device { + id: string; + packages: { publicPackage: KeyPackage; privatePackage: PrivateKeyPackage }[]; + state?: ClientState; +} + +const encodeKp = (kp: KeyPackage) => mls.encodeMlsMessage({ version: "mls10", wireformat: "mls_key_package", keyPackage: kp }); + +async function makeDevice(id: string, count: number): Promise { + const packages = []; + for (let i = 0; i < count; i++) { + packages.push( + await mls.generateKeyPackage( + { credentialType: "basic", identity: new TextEncoder().encode(id) }, + mls.defaultCapabilities(), + mls.defaultLifetime, + [], + impl, + ), + ); + } + return { id, packages }; +} + +async function publish(p: Participant, d: Device, lastResort = false): Promise { + const encoded = d.packages.map((k) => encodeKp(k.publicPackage)); + return p.call("mls:keypackages:publish", { + deviceId: d.id, + keyPackages: lastResort ? encoded.slice(0, -1) : encoded, + ...(lastResort ? { lastResort: encoded[encoded.length - 1] } : {}), + }); +} + +/* The claimed package's private half, matched by ref as a client matches it. */ +async function privateFor(d: Device, claimed: Uint8Array): Promise { + const parsed = await parseKeyPackage(claimed); + assert.ok(parsed.ok); + for (const k of d.packages) { + const mine = await parseKeyPackage(encodeKp(k.publicPackage)); + if (mine.ok && mine.ref === parsed.ref) return k; + } + throw new Error("claimed a package this device never made"); +} + +function decode(bytes: Uint8Array) { + const [msg] = mls.decodeMlsMessage(bytes, 0)!; + return msg; +} + +async function startGroup(owner: Device): Promise<{ groupIdHex: string }> { + const groupId = randomBytes(16); + owner.state = await mls.createGroup(groupId, owner.packages[0].publicPackage, owner.packages[0].privatePackage, [], impl); + return { groupIdHex: groupId.toString("hex") }; +} + +async function commitAdding(committer: Device, kps: Uint8Array[]) { + const extraProposals = kps.map((b) => { + const m = decode(b); + assert.equal(m.wireformat, "mls_key_package"); + return { proposalType: "add" as const, add: { keyPackage: (m as { keyPackage: KeyPackage }).keyPackage } }; + }); + const result = await mls.createCommit( + { state: committer.state!, cipherSuite: impl }, + { extraProposals, wireAsPublicMessage: true, ratchetTreeExtension: true }, + ); + return { + result, + commit: mls.encodeMlsMessage(result.commit), + welcome: result.welcome ? mls.encodeMlsMessage({ version: "mls10", wireformat: "mls_welcome", welcome: result.welcome }) : undefined, + }; +} + +async function applicationMessage(sender: Device, text: string): Promise { + const r = await mls.createApplicationMessage(sender.state!, new TextEncoder().encode(text), impl); + sender.state = r.newState; + return mls.encodeMlsMessage({ version: "mls10", wireformat: "mls_private_message", privateMessage: r.privateMessage }); +} + +async function receive(d: Device, bytes: Uint8Array): Promise { + const m = decode(bytes); + if (m.wireformat !== "mls_private_message" && m.wireformat !== "mls_public_message") throw new Error(m.wireformat); + const r = await mls.processMessage(m, d.state!, mls.emptyPskIndex, () => "accept", impl); + d.state = r.newState; + return r.kind === "applicationMessage" ? new TextDecoder().decode(r.message) : null; +} + +let alice: Participant; +let bob: Participant; +let mallory: Participant; +let dm: string; + +before(async () => { + dir = mkdtempSync(join(tmpdir(), "gryt-mls-")); + process.env.DATA_DIR = dir; + await initSqlite(); + await createServerConfigIfNotExists(); + resetChannelIdCache(); + impl = await mls.getCiphersuiteImpl(mls.getCiphersuiteFromName("MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519")); + + alice = await connect("Alice"); + bob = await connect("Bob"); + mallory = await connect("Mallory"); + dm = (await openDirectConversation(alice.serverUserId, bob.serverUserId)).conversation_id; +}); + +after(() => { + delete process.env.DATA_DIR; + rmSync(dir, { recursive: true, force: true }); +}); + +describe("the wire parser", () => { + it("names a Welcome's recipient by the same ref as the KeyPackage it was built from", async () => { + const a = await makeDevice("wire-a", 1); + const b = await makeDevice("wire-b", 1); + await startGroup(a); + const { commit, welcome } = await commitAdding(a, [encodeKp(b.packages[0].publicPackage)]); + + const kp = await parseKeyPackage(encodeKp(b.packages[0].publicPackage)); + const w = parseWelcome(welcome!); + const c = await parseGroupMessage(commit); + assert.ok(kp.ok && w.ok && c.ok); + assert.deepEqual(w.recipients, [kp.ref]); + assert.deepEqual([c.kind, c.epoch, c.addedRefs], ["commit", 0, [kp.ref]]); + }); + + it("refuses junk, trailing bytes and another ciphersuite", async () => { + const good = encodeKp((await makeDevice("wire-c", 1)).packages[0].publicPackage); + assert.equal((await parseKeyPackage(new Uint8Array([1, 2, 3]))).ok, false); + assert.equal((await parseKeyPackage(new Uint8Array([...good, 0]))).ok, false, "trailing bytes nobody looked at"); + + // MLSMessage header (4 bytes), then the KeyPackage's version (2) and cipher_suite (2). + const suite2 = new Uint8Array(good); + suite2[7] = 2; + const refused = await parseKeyPackage(suite2); + assert.equal(!refused.ok && refused.error, "unsupported_ciphersuite"); + }); + + it("refuses a commit sent as PrivateMessage, since its adds can't be checked", async () => { + const a = await makeDevice("wire-d", 1); + await startGroup(a); + const r = await mls.createCommit({ state: a.state!, cipherSuite: impl }, {}); + const parsed = await parseGroupMessage(mls.encodeMlsMessage(r.commit)); + assert.equal(!parsed.ok && parsed.error, "must_be_public"); + }); +}); + +describe("KeyPackages", () => { + it("are refused from somebody not signed in", async () => { + const d = await makeDevice("anon", 1); + const reply = await new Promise((resolve) => { + void alice.handlers["mls:keypackages:publish"]({ accessToken: "not-a-token", deviceId: d.id, keyPackages: [encodeKp(d.packages[0].publicPackage)] }, resolve); + }); + assert.equal(reply.error, "unauthenticated"); + }); + + it("are refused when they aren't KeyPackages", async () => { + const reply = await alice.call("mls:keypackages:publish", { deviceId: "junk", keyPackages: [Buffer.from("hello")] }); + assert.equal(reply.error, "invalid_key_package"); + }); + + it("stop at five devices per member", async () => { + const carol = await connect("Carol"); + for (let i = 0; i < 5; i++) { + assert.equal((await publish(carol, await makeDevice(`carol-${i}`, 1))).ok, true); + } + const sixth = await publish(carol, await makeDevice("carol-5", 1)); + assert.equal(sixth.error, "too_many_devices"); + + assert.equal((await carol.call("mls:device:remove", { deviceId: "carol-0" })).ok, true); + assert.equal((await publish(carol, await makeDevice("carol-5", 1))).ok, true, "removing one frees its slot"); + }); + + it("are handed out once each, then the last-resort one", async () => { + const eve = await connect("Eve"); + const frank = await connect("Frank"); + const conv = (await openDirectConversation(eve.serverUserId, frank.serverUserId)).conversation_id; + const frankPhone = await makeDevice("frank-phone", 3); + const eveLaptop = await makeDevice("eve-laptop", 1); + assert.deepEqual(await publish(frank, frankPhone, true), { ok: true, stored: 3, unclaimed: 2, lastResort: true }); + await publish(eve, eveLaptop); + + const refs = []; + for (let i = 0; i < 4; i++) { + const r = await eve.call("mls:keypackages:claim", { conversationId: conv, deviceId: "eve-laptop" }); + assert.equal(r.ok, true); + const [kp] = r.keyPackages as { keyPackage: Buffer; lastResort: boolean }[]; + const parsed = await parseKeyPackage(kp.keyPackage); + assert.ok(parsed.ok); + refs.push([parsed.ref, kp.lastResort]); + } + assert.notEqual(refs[0][0], refs[1][0], "one package to two groups would share an init key"); + assert.deepEqual(refs.map((r) => r[1]), [false, false, true, true]); + assert.equal(refs[2][0], refs[3][0]); + }); + + it("are not handed to somebody outside the conversation", async () => { + await publish(mallory, await makeDevice("mallory-1", 1)); + const r = await mallory.call("mls:keypackages:claim", { conversationId: dm, deviceId: "mallory-1" }); + assert.equal(r.error, "not_found", "the same answer as a conversation that doesn't exist"); + }); + + it("follow blocks, contact settings and the server's DM switch", async () => { + const g = await connect("Gina"); + const h = await connect("Hal"); + const conv = (await openDirectConversation(g.serverUserId, h.serverUserId)).conversation_id; + await publish(g, await makeDevice("gina-1", 1)); + await publish(h, await makeDevice("hal-1", 3)); + const claim = () => g.call("mls:keypackages:claim", { conversationId: conv, deviceId: "gina-1" }); + + await blockUser(h.grytUserId, g.grytUserId); + assert.equal((await claim()).error, "unknown_member"); + await unblockUser(h.grytUserId, g.grytUserId); + + await setContactPrefs(h.grytUserId, { messages: "nobody", calls: "nobody" }); + assert.equal((await claim()).error, "contact_refused"); + await setContactPrefs(h.grytUserId, { messages: "everyone", calls: "friends" }); + + await updateServerConfig({ allowDms: false }); + assert.equal((await claim()).error, "dms_disabled"); + await updateServerConfig({ allowDms: true }); + + assert.equal((await claim()).ok, true); + }); + + it("are rate limited, since each claim uses up somebody else's", async () => { + const i = await connect("Ida"); + const j = await connect("Jon"); + const conv = (await openDirectConversation(i.serverUserId, j.serverUserId)).conversation_id; + await publish(i, await makeDevice("ida-1", 1)); + await publish(j, await makeDevice("jon-1", 1), true); + const replies = []; + for (let n = 0; n < 25; n++) replies.push(await i.call("mls:keypackages:claim", { conversationId: conv, deviceId: "ida-1" })); + assert.ok(replies.some((r) => r.error === "rate_limited")); + }); +}); + +describe("a DM over MLS", () => { + const a1: Device = { id: "alice-laptop", packages: [] }; + const b1: Device = { id: "bob-phone", packages: [] }; + let groupIdHex: string; + + it("starts with the first group registered, and the second told to wait", async () => { + Object.assign(a1, await makeDevice(a1.id, 2)); + Object.assign(b1, await makeDevice(b1.id, 2)); + await publish(alice, a1); + await publish(bob, b1); + + ({ groupIdHex } = await startGroup(a1)); + const first = await alice.call("mls:group:create", { conversationId: dm, groupId: groupIdHex }); + const second = await bob.call("mls:group:create", { conversationId: dm, groupId: randomBytes(16).toString("hex") }); + assert.equal(first.ok, true); + assert.equal(second.error, "group_exists"); + assert.equal((second.group as { groupId: string }).groupId, groupIdHex); + }); + + it("is refused to somebody outside it", async () => { + assert.equal((await mallory.call("mls:group:create", { conversationId: dm, groupId: "ab" })).error, "not_found"); + assert.equal((await mallory.call("mls:log:fetch", { conversationId: dm, after: 0 })).error, "not_found"); + assert.equal((await mallory.call("mls:devices", { conversationId: dm })).error, "not_found"); + }); + + it("adds Bob's phone with a commit and hands it the Welcome", async () => { + const claim = await alice.call("mls:keypackages:claim", { conversationId: dm, deviceId: a1.id }); + const [kp] = claim.keyPackages as { keyPackage: Buffer; deviceId: string }[]; + assert.equal(kp.deviceId, b1.id); + + bob.clear(); + const { result, commit, welcome } = await commitAdding(a1, [kp.keyPackage]); + const reply = await alice.call("mls:commit", { conversationId: dm, deviceId: a1.id, commit, welcome }); + assert.deepEqual(reply, { ok: true, seq: 1, epoch: 1 }); + a1.state = result.newState; + + const [pushed] = bob.received("mls:welcome"); + assert.equal(pushed.deviceId, b1.id); + const sync = await bob.call("mls:sync", { deviceId: b1.id }); + const [waiting] = sync.welcomes as { welcomeId: string; data: Buffer }[]; + assert.equal(waiting.welcomeId, pushed.welcomeId); + + const w = decode(waiting.data); + if (w.wireformat !== "mls_welcome") throw new Error(w.wireformat); + const mine = await privateFor(b1, kp.keyPackage); + b1.state = await mls.joinGroup(w.welcome, mine.publicPackage, mine.privatePackage, mls.emptyPskIndex, impl); + + assert.deepEqual(await bob.call("mls:welcome:ack", { deviceId: b1.id, welcomeIds: [waiting.welcomeId] }), { ok: true, deleted: 1 }); + assert.deepEqual((await bob.call("mls:sync", { deviceId: b1.id })).welcomes, []); + }); + + it("carries a message that still opens on the other side", async () => { + alice.clear(); + bob.clear(); + const bytes = await applicationMessage(a1, "hello bob"); + const reply = await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: bytes }); + assert.deepEqual(reply, { ok: true, seq: 2 }); + + const [live] = bob.received("mls:message"); + assert.equal(live.kind, "application"); + assert.equal(await receive(b1, live.data as Buffer), "hello bob"); + assert.equal(alice.received("mls:message").length, 1, "the sender's other sockets see it too"); + assert.equal(bob.received("dm:opened").length, 1, "the first message puts the conversation in Bob's list"); + }); + + it("tells the other side about a new device, and lists devices by conversation", async () => { + alice.clear(); + const b2 = await makeDevice("bob-laptop", 1); + await publish(bob, b2); + assert.deepEqual(alice.received("mls:devices:changed"), [{ serverUserId: bob.serverUserId }]); + + const listed = await alice.call("mls:devices", { conversationId: dm }); + const ids = (listed.devices as { deviceId: string }[]).map((d) => d.deviceId).sort(); + assert.deepEqual(ids, ["alice-laptop", "bob-laptop", "bob-phone"]); + const own = await bob.call("mls:devices"); + assert.equal((own.devices as unknown[]).length, 2); + + const sync = await bob.call("mls:sync", { deviceId: "bob-laptop" }); + assert.deepEqual(sync.keyPackages, { unclaimed: 1, lastResort: false, target: 20 }); + assert.deepEqual((sync.groups as { conversationId: string }[]).map((g) => g.conversationId), [dm]); + assert.equal((await bob.call("mls:device:remove", { deviceId: "bob-laptop" })).ok, true); + }); + + it("orders two commits racing for one epoch", async () => { + const [fromAlice, fromBob] = await Promise.all([ + mls.createCommit({ state: a1.state!, cipherSuite: impl }, { wireAsPublicMessage: true }), + mls.createCommit({ state: b1.state!, cipherSuite: impl }, { wireAsPublicMessage: true }), + ]); + const [ra, rb] = await Promise.all([ + alice.call("mls:commit", { conversationId: dm, deviceId: a1.id, commit: mls.encodeMlsMessage(fromAlice.commit) }), + bob.call("mls:commit", { conversationId: dm, deviceId: b1.id, commit: mls.encodeMlsMessage(fromBob.commit) }), + ]); + + const replies = [ra, rb]; + assert.equal(replies.filter((r) => r.ok).length, 1, "exactly one commit per epoch"); + const loser = replies.find((r) => !r.ok)!; + assert.deepEqual([loser.error, loser.epoch], ["stale_epoch", 2]); + + // The loser catches up from its cursor, as a client would, and both agree again. + const aliceWon = ra.ok; + const winnerState = aliceWon ? fromAlice.newState : fromBob.newState; + const behind = aliceWon ? b1 : a1; + const fetched = await (aliceWon ? bob : alice).call("mls:log:fetch", { conversationId: dm, after: 2 }); + const [entry] = fetched.entries as { kind: string; data: Buffer }[]; + assert.equal(entry.kind, "commit"); + await receive(behind, entry.data); + if (aliceWon) a1.state = winnerState; + else b1.state = winnerState; + assert.equal(a1.state!.groupContext.epoch, b1.state!.groupContext.epoch); + + const retried = await (aliceWon ? bob : alice).call("mls:commit", { + conversationId: dm, + deviceId: behind.id, + commit: mls.encodeMlsMessage((aliceWon ? fromBob : fromAlice).commit), + }); + assert.equal(retried.error, "stale_epoch", "a commit built on the old epoch stays refused"); + }); + + it("refuses a message from an epoch that hasn't happened, and one for another group", async () => { + const other = await makeDevice("other", 1); + await startGroup(other); + const foreign = await applicationMessage(other, "wrong group"); + assert.equal((await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: foreign })).error, "wrong_group"); + + const commit = await mls.createCommit({ state: a1.state!, cipherSuite: impl }, { wireAsPublicMessage: true }); + const ahead = await applicationMessage({ ...a1, state: commit.newState }, "from the future"); + assert.equal((await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: ahead })).error, "future_epoch"); + }); + + it("won't add a device that belongs to nobody in the conversation", async () => { + const outsider = await makeDevice("mallory-2", 1); + await publish(mallory, outsider); + const { commit, welcome } = await commitAdding(a1, [encodeKp(outsider.packages[0].publicPackage)]); + const r = await alice.call("mls:commit", { conversationId: dm, deviceId: a1.id, commit, welcome }); + assert.equal(r.error, "not_a_member_device"); + }); + + it("pages the log from a cursor", async () => { + for (let i = 0; i < 3; i++) { + await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: await applicationMessage(a1, `m${i}`) }); + } + const first = await bob.call("mls:log:fetch", { conversationId: dm, after: 0, limit: 2 }); + assert.deepEqual((first.entries as { seq: number }[]).map((e) => e.seq), [1, 2]); + assert.equal(first.hasMore, true); + const rest = await bob.call("mls:log:fetch", { conversationId: dm, after: first.nextCursor as number, limit: 50 }); + assert.equal(rest.hasMore, false); + assert.equal(rest.gap, false); + const all = [...(first.entries as { seq: number }[]), ...(rest.entries as { seq: number }[])].map((e) => e.seq); + assert.deepEqual(all, all.map((_, i) => i + 1), "no seq skipped, none repeated"); + }); + + it("holds back a blocked sender's messages but never a commit", async () => { + await blockUser(bob.grytUserId, alice.grytUserId); + bob.clear(); + const head = (getMlsGroupForConversation(dm)!).headSeq; + + await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: await applicationMessage(a1, "ignored") }); + const c = await mls.createCommit({ state: a1.state!, cipherSuite: impl }, { wireAsPublicMessage: true }); + const cr = await alice.call("mls:commit", { conversationId: dm, deviceId: a1.id, commit: mls.encodeMlsMessage(c.commit) }); + assert.equal(cr.ok, true); + a1.state = c.newState; + + assert.deepEqual(bob.received("mls:message").map((m) => m.kind), ["commit"]); + const fetched = await bob.call("mls:log:fetch", { conversationId: dm, after: head }); + assert.deepEqual((fetched.entries as { kind: string }[]).map((e) => e.kind), ["commit"]); + assert.equal(fetched.nextCursor, head + 2, "the cursor still moves past what was held back"); + await unblockUser(bob.grytUserId, alice.grytUserId); + }); + + it("refuses a muted member's message, as chat:send does", async () => { + await setUserModerationState(alice.serverUserId, { muted: true, mutedUntil: new Date(Date.now() + 60_000) }); + const r = await alice.call("mls:send", { conversationId: dm, deviceId: a1.id, message: await applicationMessage(a1, "muted") }); + assert.equal(r.error, "muted"); + await setUserModerationState(alice.serverUserId, { muted: false, mutedUntil: null }); + }); + + it("isn't offered for a group DM yet", async () => { + const group = await createGroupConversation(alice.serverUserId, [bob.serverUserId, mallory.serverUserId]); + const r = await alice.call("mls:group:create", { conversationId: group.conversation_id, groupId: "cd" }); + assert.equal(r.error, "not_supported"); + }); +}); + +describe("retention", () => { + it("is 30 days, and a host can only make it shorter", () => { + assert.equal(mlsRetentionDays({}), 30); + assert.equal(mlsRetentionDays({ MLS_RETENTION_DAYS: "7" }), 7); + assert.equal(mlsRetentionDays({ MLS_RETENTION_DAYS: "90" }), 30); + assert.equal(mlsRetentionDays({ MLS_RETENTION_DAYS: "0" }), 30); + assert.equal(mlsRetentionDays({ MLS_RETENTION_DAYS: "soon" }), 30); + }); + + 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(); + getSqliteDb().prepare(`UPDATE mls_log SET created_at = ? WHERE group_id = ? AND seq <= 3`).run(old, group.groupId); + + process.env.MLS_RETENTION_DAYS = "7"; + try { + assert.ok(runMlsRetention().log >= 3); + } finally { + delete process.env.MLS_RETENTION_DAYS; + } + + const behind = await bob.call("mls:log:fetch", { conversationId: dm, after: 1 }); + assert.equal(behind.gap, true); + assert.equal((behind.entries as { seq: number }[])[0].seq, 4); + const caughtUp = await bob.call("mls:log:fetch", { conversationId: dm, after: 3 }); + assert.equal(caughtUp.gap, false); + }); +}); diff --git a/src/socket/handlers/mls.ts b/src/socket/handlers/mls.ts new file mode 100644 index 00000000..dc4be578 --- /dev/null +++ b/src/socket/handlers/mls.ts @@ -0,0 +1,709 @@ +import consola from "consola"; + +import { + addMlsKeyPackages, + appendMlsCommit, + appendMlsMessage, + blockedServerIdsFor, + blockersOfSender, + claimMlsKeyPackage, + countMlsKeyPackages, + createMlsGroup, + deleteMlsWelcomes, + eitherHasBlocked, + getConversation, + getMlsGroupForConversation, + getServerConfig, + getUserByServerId, + isMlsDevice, + listConversationsForUser, + listMlsDevices, + listMlsGroupsForMember, + listMlsLog, + listMlsWelcomes, + MLS_MAX_KEY_PACKAGES, + mlsKeyPackageOwner, + oldestMlsSeq, + removeMlsDevice, + touchConversation, + touchMlsDevice, + type MlsGroup, + type MlsLogEntry, + type MlsWelcomeFor, +} from "../../db"; +import { spamFilter } from "../../moderation/spamFilter"; +import { isSpamExempt, spamRefusal, timeOutSpammer } from "../../moderation/spamTimeout"; +import { textMuteError, textMuteFor } from "../../moderation/textMute"; +import { mayInChannel } from "../../services/channelPermissions"; +import { asBytes, parseGroupMessage, parseKeyPackage, parseWelcome } from "../../services/mlsWire"; +import { SEALED_MAX_LENGTH } from "../../utils/messageLimits"; +import { checkRateLimit, type RateLimitRule } from "../../utils/rateLimiter"; +import { requireAuth, type AuthResult } from "../middleware/auth"; +import { CONTACT_REFUSALS, mayMessage, peerOf } from "../utils/contactGate"; +import { DENIAL_RESPONSES, resolveConversationAccess } from "../utils/conversationAccess"; +import { socketIsIdentified } from "../utils/standing"; +import { directConversationViews } from "./dm"; +import type { EventHandlerMap, HandlerContext } from "./types"; + +/** + * The MLS delivery service (GRYT-1500, stage 1 of docs/mls-design.md in the crypto repo). + * Every request answers through its ack as `{ ok: true, ... }` or `{ ok: false, error, message }`. + */ + +type Reply = Record & { ok: boolean }; +type Ack = (reply: Reply) => void; + +const DEVICE_ID = /^[A-Za-z0-9_-]{1,64}$/; +/** 1 to 64 bytes, lower-case hex. RFC 9420 leaves the length to the application. */ +const GROUP_ID = /^(?:[0-9a-f]{2}){1,64}$/; + +/** A suite 1 KeyPackage is about 330 bytes plus the device certificate in its credential. */ +const MAX_KEY_PACKAGE_BYTES = 16 * 1024; +/** A DM commit adds at most ten devices; a Welcome carries the tree for all of them. */ +const MAX_HANDSHAKE_BYTES = 256 * 1024; +const MAX_LOG_PAGE = 200; +const MAX_CLAIM_DEVICES = 50; + +const RL_PUBLISH: RateLimitRule = { limit: 10, windowMs: 60_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 6000 }; +/** Every claim uses up somebody else's packages, so it is the one to hold back. */ +const RL_CLAIM: RateLimitRule = { limit: 20, windowMs: 60_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 3000 }; +const RL_COMMIT: RateLimitRule = { limit: 30, windowMs: 60_000, scorePerAction: 1, maxScore: 15, scoreDecayMs: 2000 }; +/** chat:send's numbers, under its own key so MLS and sealed sends don't share one budget. */ +const RL_SEND: RateLimitRule = { limit: 20, windowMs: 10_000, banMs: 30_000, scorePerAction: 1, maxScore: 10, scoreDecayMs: 2000 }; +const RL_READ: RateLimitRule = { limit: 120, windowMs: 60_000, scorePerAction: 0.2, maxScore: 20, scoreDecayMs: 500 }; + +const fail = (error: string, message: string, extra: Record = {}): Reply => ({ + ok: false, + error, + message, + ...extra, +}); + +function groupView(g: MlsGroup): Record { + return { conversationId: g.conversationId, groupId: g.groupId, epoch: g.epoch, headSeq: g.headSeq }; +} + +function entryView(conversationId: string, e: MlsLogEntry): Record { + return { + conversationId, + groupId: e.groupId, + seq: e.seq, + kind: e.kind, + epoch: e.epoch, + senderServerUserId: e.senderServerUserId, + senderDeviceId: e.senderDeviceId, + data: Buffer.from(e.data), + createdAt: e.createdAt, + }; +} + +export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { + const { io, socket, clientId, clientsInfo, serverId, sfuClient, getClientIp } = ctx; + + function socketsOf(serverUserIds: Iterable): string[] { + const wanted = new Set(serverUserIds); + return Object.entries(clientsInfo) + .filter(([cid, ci]) => socketIsIdentified(clientsInfo, cid) && wanted.has(ci.serverUserId)) + .map(([cid]) => cid); + } + + function emitTo(socketIds: string[], event: string, payload: unknown): void { + for (const cid of socketIds) io.sockets.sockets.get(cid)?.emit(event, payload); + } + + /* Rate limit, then the token. Null once the refusal has gone back through the ack. */ + async function begin( + event: string, + rule: RateLimitRule, + payload: { accessToken?: string } | undefined, + ack: Ack, + ): Promise { + const rl = checkRateLimit(event, clientsInfo[clientId]?.serverUserId, getClientIp(), rule); + if (!rl.allowed) { + ack(fail("rate_limited", `Too fast. Wait ${Math.ceil((rl.retryAfterMs || 0) / 1000)}s.`, { retryAfterMs: rl.retryAfterMs })); + return null; + } + if (!payload || typeof payload !== "object") { + ack(fail("invalid_payload", "Invalid payload")); + return null; + } + const auth = await requireAuth(socket, payload); + if (!auth) ack(fail("unauthenticated", "Sign in to this server first.")); + return auth; + } + + /* One-to-one DMs only in stage 1; group DMs are stage 2 and bring their own checks. */ + async function dmOf(conversationId: unknown, serverUserId: string, ack: Ack): Promise { + if (typeof conversationId !== "string" || !conversationId) { + ack(fail("invalid_payload", "conversationId is required.")); + return null; + } + const access = await resolveConversationAccess(conversationId, serverUserId); + if (!access.allowed) { + const { error, message } = DENIAL_RESPONSES[access.reason]; + ack(fail(error, message)); + return null; + } + if (access.kind !== "dm" || access.group) { + ack(fail("not_supported", "Only one-to-one direct messages use MLS so far.")); + return null; + } + return access.memberIds; + } + + function ownDevice(auth: AuthResult, deviceId: unknown, ack: Ack, mustExist = true): deviceId is string { + if (typeof deviceId !== "string" || !DEVICE_ID.test(deviceId)) { + ack(fail("invalid_device", "deviceId has to be 1 to 64 letters, digits, - or _.")); + return false; + } + if (mustExist && !isMlsDevice(auth.tokenPayload.serverUserId, deviceId)) { + ack(fail("unknown_device", "Publish KeyPackages from this device first.")); + return false; + } + return true; + } + + /* What chat:send asks before a DM goes out, minus what needs the text. Null is yes. */ + async function sendRefusal(auth: AuthResult, conversationId: string, memberIds: string[]): Promise { + const self = auth.tokenPayload.serverUserId; + const mute = await textMuteFor(self); + if (mute.muted) return { ok: false, ...textMuteError(mute) }; + if (!(await mayInChannel(conversationId, self, "send_messages", auth.tokenPayload.grytUserId))) { + return fail("forbidden", "You don't have permission to send messages on this server.", { permission: "send_messages" }); + } + return writeRefusal(auth, memberIds); + } + + /* Anything that reaches the other person: a message, or taking their KeyPackages. */ + async function writeRefusal(auth: AuthResult, memberIds: string[]): Promise { + const self = auth.tokenPayload.serverUserId; + const cfg = await getServerConfig().catch(() => null); + if (cfg && cfg.allow_dms === false) return fail("dms_disabled", "Direct messages are turned off on this server"); + if (!auth.permissions.has("send_direct_messages")) { + return fail("forbidden", "You do not have permission to send direct messages here.", { permission: "send_direct_messages" }); + } + const peer = peerOf(memberIds, self); + if (peer && !(await mayMessage(self, peer))) return { ok: false, ...CONTACT_REFUSALS.messages }; + return null; + } + + /* 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 { + 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; + const user = await getUserByServerId(auth.tokenPayload.serverUserId); + if (!user) return fail("unknown_member", "User not found. Please rejoin."); + + const sensitivity = cfg?.spam_filter_sensitivity ?? "normal"; + const verdict = spamFilter.evaluate( + { id: user.server_user_id, memberSince: user.created_at }, + { + kind: "dm", + conversationId, + recipients: memberIds.filter((id) => id !== user.server_user_id), + size, + newConversation, + attachments: 0, + }, + sensitivity, + ); + if (!verdict.spam) return null; + const { until } = await timeOutSpammer({ + io, clientsInfo, sfuClient, serverId, serverUserId: user.server_user_id, verdict, sensitivity, where: "dm", + }); + return { ok: false, ...spamRefusal(until) }; + } + + /* 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); + const owners = []; + for (const ref of refs) { + const owner = mlsKeyPackageOwner(ref); + if (!owner || !members.has(owner.serverUserId)) return null; + owners.push(owner); + } + return owners; + } + + /* Commits always arrive, or the reader's group state falls behind. Anything else + skips people who blocked the sender, as chat:send does. */ + async function fanOut(memberIds: string[], entry: MlsLogEntry, conversationId: string): Promise { + let audience = memberIds; + if (entry.kind !== "commit") { + const blockers = await blockersOfSender(entry.senderServerUserId); + audience = memberIds.filter((id) => id === entry.senderServerUserId || !blockers.has(id)); + } + emitTo(socketsOf(audience), "mls:message", entryView(conversationId, entry)); + } + + /* Whoever shares a DM with them adds the new device the next time they send. */ + async function announceDevices(serverUserId: string): Promise { + const people = new Set([serverUserId]); + for (const c of await listConversationsForUser(serverUserId)) { + for (const id of c.other_server_user_ids) people.add(id); + } + emitTo(socketsOf(people), "mls:devices:changed", { serverUserId }); + } + + return { + "mls:keypackages:publish": async ( + payload: { accessToken: string; deviceId: string; keyPackages?: unknown[]; lastResort?: unknown }, + ack: Ack, + ) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:keypackages:publish", RL_PUBLISH, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack, false)) return; + const self = auth.tokenPayload.serverUserId; + + const regular = Array.isArray(payload.keyPackages) ? payload.keyPackages : []; + const raw = [...regular.map((b) => ({ b, lastResort: false }))]; + if (payload.lastResort !== undefined && payload.lastResort !== null) raw.push({ b: payload.lastResort, lastResort: true }); + if (raw.length === 0 || regular.length > MLS_MAX_KEY_PACKAGES) { + ack(fail("invalid_payload", `Send 1 to ${MLS_MAX_KEY_PACKAGES} KeyPackages, plus a last-resort one if you like.`)); + return; + } + + const packages = []; + for (const { b, lastResort } of raw) { + const bytes = asBytes(b); + if (!bytes || bytes.length > MAX_KEY_PACKAGE_BYTES) { + ack(fail("invalid_key_package", "Each KeyPackage has to be binary and under 16 kB.")); + return; + } + const parsed = await parseKeyPackage(bytes); + if (!parsed.ok) { + ack(fail(parsed.error, parsed.message)); + return; + } + packages.push({ ref: parsed.ref, data: bytes, lastResort }); + } + + const isNew = !isMlsDevice(self, payload.deviceId); + if (touchMlsDevice(self, payload.deviceId) === "too_many_devices") { + ack(fail("too_many_devices", "You have five devices using encrypted messages here. Remove one first.")); + return; + } + const result = addMlsKeyPackages(self, payload.deviceId, packages); + ack({ ok: true, ...result }); + if (isNew) await announceDevices(self); + } catch (err) { + consola.error("mls:keypackages:publish failed", err); + ack(fail("failed", "Could not store the KeyPackages")); + } + }, + + /** Your own devices with no conversation named; with one, every member's device ids. */ + "mls:devices": async (payload: { accessToken: string; conversationId?: string }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:read", RL_READ, payload, ack); + if (!auth) return; + const self = auth.tokenPayload.serverUserId; + if (payload.conversationId === undefined) { + ack({ ok: true, devices: listMlsDevices([self]) }); + return; + } + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + const devices = listMlsDevices(memberIds).map((d) => ({ serverUserId: d.serverUserId, deviceId: d.deviceId })); + ack({ ok: true, devices }); + } catch (err) { + consola.error("mls:devices failed", err); + ack(fail("failed", "Could not list devices")); + } + }, + + "mls:device:remove": async (payload: { accessToken: string; deviceId: string }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:keypackages:publish", RL_PUBLISH, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack)) return; + removeMlsDevice(auth.tokenPayload.serverUserId, payload.deviceId); + ack({ ok: true }); + await announceDevices(auth.tokenPayload.serverUserId); + } catch (err) { + consola.error("mls:device:remove failed", err); + ack(fail("failed", "Could not remove the device")); + } + }, + + /** + * One KeyPackage for each device named, or for every device in the conversation + * except the caller's. Handed out once; a device with none left is listed as missing. + */ + "mls:keypackages:claim": async ( + payload: { accessToken: string; conversationId: string; deviceId: string; devices?: { serverUserId: string; deviceId: string }[] }, + ack: Ack, + ) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:keypackages:claim", RL_CLAIM, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack)) return; + const self = auth.tokenPayload.serverUserId; + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + + const known = listMlsDevices(memberIds); + let targets = known.filter((d) => !(d.serverUserId === self && d.deviceId === payload.deviceId)); + if (payload.devices !== undefined) { + if (!Array.isArray(payload.devices)) { + ack(fail("invalid_payload", "devices has to be a list.")); + return; + } + const asked = new Set(payload.devices.map((d) => `${d?.serverUserId}\u0000${d?.deviceId}`)); + targets = targets.filter((d) => asked.has(`${d.serverUserId}\u0000${d.deviceId}`)); + } + targets = targets.slice(0, MAX_CLAIM_DEVICES); + + const peers = new Set(targets.map((d) => d.serverUserId).filter((id) => id !== self)); + if (peers.size > 0) { + const refusal = await writeRefusal(auth, memberIds); + if (refusal) { + ack(refusal); + return; + } + for (const peer of peers) { + const user = await getUserByServerId(peer); + // As dm:open answers it, so a block reads the same as somebody who isn't there. + if (!user || (await eitherHasBlocked(auth.tokenPayload.grytUserId, user.gryt_user_id))) { + ack(fail("unknown_member", "That person is not a member of this server")); + return; + } + } + } + + const keyPackages = []; + const missing = []; + for (const d of targets) { + const kp = claimMlsKeyPackage(d.serverUserId, d.deviceId); + if (kp) { + keyPackages.push({ serverUserId: d.serverUserId, deviceId: d.deviceId, keyPackage: Buffer.from(kp.data), lastResort: kp.lastResort }); + } else { + missing.push({ serverUserId: d.serverUserId, deviceId: d.deviceId }); + } + } + ack({ ok: true, keyPackages, missing }); + } catch (err) { + consola.error("mls:keypackages:claim failed", err); + ack(fail("failed", "Could not hand out KeyPackages")); + } + }, + + /** The first group for a conversation wins. The loser drops its own and waits for a Welcome. */ + "mls:group:create": async (payload: { accessToken: string; conversationId: string; groupId: string }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:commit", RL_COMMIT, payload, ack); + if (!auth) return; + const self = auth.tokenPayload.serverUserId; + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + if (typeof payload.groupId !== "string" || !GROUP_ID.test(payload.groupId)) { + ack(fail("invalid_group_id", "groupId has to be 1 to 64 bytes of lower-case hex.")); + return; + } + + const result = createMlsGroup(payload.groupId, payload.conversationId, self); + if (result.created) ack({ ok: true, group: groupView(result.group) }); + else if (result.group) ack(fail("group_exists", "This conversation already has a group.", { group: groupView(result.group) })); + else ack(fail("group_id_taken", "Pick another group id.")); + } catch (err) { + consola.error("mls:group:create failed", err); + ack(fail("failed", "Could not create the group")); + } + }, + + /** + * A commit built on the group's current epoch, with the Welcome for anybody it adds. + * Any other epoch is refused with the current one, and the client catches up and retries. + */ + "mls:commit": async ( + payload: { accessToken: string; conversationId: string; deviceId: string; commit: unknown; welcome?: unknown }, + ack: Ack, + ) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:commit", RL_COMMIT, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack)) return; + const self = auth.tokenPayload.serverUserId; + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + const group = getMlsGroupForConversation(payload.conversationId); + if (!group) { + ack(fail("no_group", "This conversation has no group yet.")); + return; + } + + const commit = asBytes(payload.commit); + const welcome = payload.welcome == null ? null : asBytes(payload.welcome); + if (!commit || commit.length > MAX_HANDSHAKE_BYTES || (payload.welcome != null && (!welcome || welcome.length > MAX_HANDSHAKE_BYTES))) { + ack(fail("invalid_payload", "commit and welcome have to be binary and under 256 kB.")); + return; + } + + const parsed = await parseGroupMessage(commit); + if (!parsed.ok) { + ack(fail(parsed.error, parsed.message)); + return; + } + if (parsed.kind !== "commit") { + ack(fail("not_a_commit", "Send proposals and application messages with mls:send.")); + return; + } + if (parsed.groupId !== group.groupId) { + ack(fail("wrong_group", "That commit is for another group.", { group: groupView(group) })); + return; + } + if (!memberDevicesFor(parsed.addedRefs, memberIds)) { + ack(fail("not_a_member_device", "A commit here can only add devices of people in the conversation.")); + return; + } + + const welcomes: MlsWelcomeFor[] = []; + if (welcome) { + const w = parseWelcome(welcome); + if (!w.ok) { + ack(fail(w.error, w.message)); + return; + } + const devices = memberDevicesFor(w.recipients, memberIds); + if (!devices) { + ack(fail("not_a_member_device", "That Welcome is for somebody outside the conversation.")); + return; + } + const seen = new Set(); + for (const d of devices) { + const key = `${d.serverUserId}\u0000${d.deviceId}`; + if (seen.has(key)) continue; + seen.add(key); + welcomes.push({ ...d, data: welcome }); + } + } + + const result = appendMlsCommit( + { groupId: group.groupId, epoch: parsed.epoch, senderServerUserId: self, senderDeviceId: payload.deviceId, data: commit }, + welcomes, + ); + if (!result.accepted) { + if (result.reason === "stale_epoch") { + ack(fail("stale_epoch", "Another commit got there first. Fetch the log and try again.", { epoch: result.epoch, headSeq: result.headSeq })); + } else { + ack(fail("no_group", "This conversation has no group yet.")); + } + return; + } + + const accepted = { ok: true, seq: result.seq, epoch: result.epoch }; + if (result.duplicate) { + ack(accepted); + return; + } + + const entry: MlsLogEntry = { + groupId: group.groupId, seq: result.seq, kind: "commit", epoch: parsed.epoch, + senderServerUserId: self, senderDeviceId: payload.deviceId, data: commit, createdAt: result.createdAt, + }; + await fanOut(memberIds, entry, payload.conversationId); + welcomes.forEach((w, i) => { + emitTo(socketsOf([w.serverUserId]), "mls:welcome", { + welcomeId: result.welcomeIds[i], + conversationId: payload.conversationId, + groupId: group.groupId, + deviceId: w.deviceId, + data: Buffer.from(w.data), + createdAt: result.createdAt, + }); + }); + ack(accepted); + } catch (err) { + consola.error("mls:commit failed", err); + ack(fail("failed", "Could not accept the commit")); + } + }, + + /** An application message (PrivateMessage) or a proposal (PublicMessage). */ + "mls:send": async ( + payload: { accessToken: string; conversationId: string; deviceId: string; message: unknown }, + ack: Ack, + ) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:send", RL_SEND, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack)) return; + const self = auth.tokenPayload.serverUserId; + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + const group = getMlsGroupForConversation(payload.conversationId); + if (!group) { + ack(fail("no_group", "This conversation has no group yet.")); + return; + } + + const bytes = asBytes(payload.message); + if (!bytes || bytes.length > SEALED_MAX_LENGTH) { + ack(fail("invalid_payload", "message has to be binary and under 64 kB.")); + return; + } + const parsed = await parseGroupMessage(bytes); + if (!parsed.ok) { + ack(fail(parsed.error, parsed.message)); + return; + } + if (parsed.kind === "commit") { + ack(fail("use_commit", "Send commits with mls:commit, so they're ordered.")); + return; + } + if (parsed.groupId !== group.groupId) { + ack(fail("wrong_group", "That message is for another group.", { group: groupView(group) })); + return; + } + if (!memberDevicesFor(parsed.addedRefs, memberIds)) { + ack(fail("not_a_member_device", "A proposal here can only add devices of people in the conversation.")); + return; + } + + const application = parsed.kind === "application"; + let wasEmpty = false; + if (application) { + const refusal = await sendRefusal(auth, payload.conversationId, memberIds); + if (refusal) { + ack(refusal); + return; + } + wasEmpty = !(await getConversation(payload.conversationId))?.last_message_at; + const spam = await droppedAsSpam(auth, payload.conversationId, memberIds, bytes.length, wasEmpty); + if (spam) { + ack(spam); + return; + } + } + + const result = appendMlsMessage(parsed.kind, { + groupId: group.groupId, epoch: parsed.epoch, senderServerUserId: self, senderDeviceId: payload.deviceId, data: bytes, + }); + if (!result.accepted) { + if (result.reason === "future_epoch") ack(fail("future_epoch", "That epoch hasn't happened yet.", { epoch: result.epoch })); + else ack(fail("no_group", "This conversation has no group yet.")); + return; + } + if (result.duplicate) { + ack({ ok: true, seq: result.seq }); + return; + } + + if (application) { + await touchConversation(payload.conversationId, new Date(result.createdAt)).catch((err) => + consola.warn("touchConversation failed", payload.conversationId, err), + ); + // The first message is what puts the conversation in the other person's list. + if (wasEmpty) { + for (const id of memberIds) { + const view = (await directConversationViews(id)).find((v) => v.conversation_id === payload.conversationId); + if (view) emitTo(socketsOf([id]), "dm:opened", view); + } + } + } + await fanOut(memberIds, { + groupId: group.groupId, seq: result.seq, kind: parsed.kind, epoch: parsed.epoch, + senderServerUserId: self, senderDeviceId: payload.deviceId, data: bytes, createdAt: result.createdAt, + }, payload.conversationId); + ack({ ok: true, seq: result.seq }); + } catch (err) { + consola.error("mls:send failed", err); + ack(fail("failed", "Could not send")); + } + }, + + /** + * Everything after the cursor, oldest first. `gap` says entries the cursor needed were + * already swept, so the device has lost them and has to be re-added. + */ + "mls:log:fetch": async (payload: { accessToken: string; conversationId: string; after: number; limit?: number }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:read", RL_READ, payload, ack); + if (!auth) return; + const self = auth.tokenPayload.serverUserId; + const memberIds = await dmOf(payload.conversationId, self, ack); + if (!memberIds) return; + const after = payload.after; + if (typeof after !== "number" || !Number.isSafeInteger(after) || after < 0) { + ack(fail("invalid_payload", "after has to be a seq, 0 for the start.")); + return; + } + const limit = typeof payload.limit === "number" && payload.limit >= 1 ? Math.min(Math.floor(payload.limit), MAX_LOG_PAGE) : 100; + const group = getMlsGroupForConversation(payload.conversationId); + if (!group) { + ack({ ok: true, group: null, entries: [], nextCursor: after, hasMore: false, gap: false }); + return; + } + + const rows = listMlsLog(group.groupId, after, limit + 1); + const page = rows.slice(0, limit); + const blocked = await blockedServerIdsFor(self); + const entries = page + .filter((e) => e.kind === "commit" || !blocked.has(e.senderServerUserId)) + .map((e) => entryView(payload.conversationId, e)); + const oldest = oldestMlsSeq(group.groupId); + const gap = after < group.headSeq && (oldest === null || oldest > after + 1); + ack({ + ok: true, + group: groupView(group), + entries, + nextCursor: page.length > 0 ? page[page.length - 1].seq : Math.max(after, gap ? group.headSeq : after), + hasMore: rows.length > limit, + gap, + }); + } catch (err) { + consola.error("mls:log:fetch failed", err); + ack(fail("failed", "Could not read the log")); + } + }, + + /** What a device needs on connect: its groups' heads, its Welcomes, and its package count. */ + "mls:sync": async (payload: { accessToken: string; deviceId: string }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:read", RL_READ, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack, false)) return; + const self = auth.tokenPayload.serverUserId; + const registered = isMlsDevice(self, payload.deviceId); + if (registered) touchMlsDevice(self, payload.deviceId); + + ack({ + ok: true, + registered, + groups: listMlsGroupsForMember(self).map((g) => ({ ...groupView(g), oldestSeq: g.oldestSeq })), + welcomes: registered + ? listMlsWelcomes(self, payload.deviceId).map((w) => ({ ...w, data: Buffer.from(w.data) })) + : [], + keyPackages: { ...countMlsKeyPackages(self, payload.deviceId), target: MLS_MAX_KEY_PACKAGES }, + }); + } catch (err) { + consola.error("mls:sync failed", err); + ack(fail("failed", "Could not sync")); + } + }, + + /** Once a Welcome is processed and saved. Until then it stays, so a crash loses nothing. */ + "mls:welcome:ack": async (payload: { accessToken: string; deviceId: string; welcomeIds: string[] }, ack: Ack) => { + ack = typeof ack === "function" ? ack : () => {}; + try { + const auth = await begin("mls:read", RL_READ, payload, ack); + if (!auth || !ownDevice(auth, payload.deviceId, ack)) return; + if (!Array.isArray(payload.welcomeIds) || payload.welcomeIds.length > 100) { + ack(fail("invalid_payload", "welcomeIds has to be a list of up to 100.")); + return; + } + const ids = payload.welcomeIds.filter((id): id is string => typeof id === "string"); + ack({ ok: true, deleted: deleteMlsWelcomes(auth.tokenPayload.serverUserId, payload.deviceId, ids) }); + } catch (err) { + consola.error("mls:welcome:ack failed", err); + ack(fail("failed", "Could not delete the Welcomes")); + } + }, + }; +} diff --git a/src/socket/index.ts b/src/socket/index.ts index 953954b5..8dc24cf4 100644 --- a/src/socket/index.ts +++ b/src/socket/index.ts @@ -50,6 +50,7 @@ import { registerFriendHandlers } from "./handlers/friends"; import { registerTypingHandlers } from "./handlers/typing"; import { registerPluginHandlers } from "./handlers/plugins"; import { registerDmKeyHandlers } from "./handlers/dmKeys"; +import { registerMlsHandlers } from "./handlers/mls"; import { registerMentionHandlers } from "./handlers/mentions"; import { addressIsOwn, resolveClientIp, trustedProxyHops } from "../config/clientAddress"; @@ -346,6 +347,7 @@ export function socketHandler(io: Server, socket: Socket, sfuClient: SFUClient | ...registerTypingHandlers(ctx), ...registerPluginHandlers(ctx), ...registerDmKeyHandlers(ctx), + ...registerMlsHandlers(ctx), ...registerMentionHandlers(ctx), }; diff --git a/src/socket/utils/server.ts b/src/socket/utils/server.ts index c17de2e4..0ac5abea 100644 --- a/src/socket/utils/server.ts +++ b/src/socket/utils/server.ts @@ -32,6 +32,8 @@ import { } from "../../db"; import { pushVoiceCapabilities } from "./voiceCapabilities"; +import { MLS_CIPHERSUITE_ID } from "../../services/mlsWire"; +import { mlsRetentionDays } from "../../jobs/mlsRetention"; // Module-level references set by socketHandler so REST routes can trigger broadcasts let _io: Server | null = null; @@ -213,6 +215,9 @@ export async function sendInfo(socket: Socket, clientsInfo: Clients | undefined, /** A constant, not a setting: either the code relays the column or it does not. Absent means no, so no client needs a table of versions. */ encryptedDirectMessages: true, + /** The MLS delivery service (GRYT-1500). A client only uses MLS where this is present, + and `retentionDays` is how long a device can be away and still catch up. */ + mls: { version: 1, ciphersuites: [MLS_CIPHERSUITE_ID], retentionDays: mlsRetentionDays() }, }; socket.emit("server:info", serverInfo); diff --git a/src/socket/utils/serverInfo.test.ts b/src/socket/utils/serverInfo.test.ts index f4bcdf96..98d97757 100644 --- a/src/socket/utils/serverInfo.test.ts +++ b/src/socket/utils/serverInfo.test.ts @@ -90,3 +90,13 @@ describe("server:info version disclosure", () => { assert.ok(!("version" in emitted[0].payload)); }); }); + +describe("server:info MLS capability", () => { + it("advertises the delivery service with its suite and retention", async () => { + const emitted: Emitted[] = []; + await sendInfo(fakeSocket(emitted), { [SOCKET_ID]: client({ serverUserId: "temp_1" }) }, "server-1"); + + assert.deepEqual(emitted[0].payload.mls, { version: 1, ciphersuites: [1], retentionDays: 30 }, + "a client only uses MLS on a server that says so"); + }); +}); diff --git a/yarn.lock b/yarn.lock index 5cd8e016..5ee534f2 100644 --- a/yarn.lock +++ b/yarn.lock @@ -523,6 +523,18 @@ resolved "https://registry.yarnpkg.com/@exodus/bytes/-/bytes-1.15.1.tgz#b13bc464ca162c17abf0837fb3a11aeab79e45d1" integrity sha512-S6mL0yNB/Abt9Ei4tq8gDhcczc4S3+vQ4ra7vxnAf+YHC02srtqxKKZghx2Dq6p0e66THKwR6r8N6P95wEty7Q== +"@hpke/common@^1.10.0": + version "1.10.1" + resolved "https://registry.yarnpkg.com/@hpke/common/-/common-1.10.1.tgz#11f205e5ba24d558c1bd4ac671580d95d488af18" + integrity sha512-moJwhmtLtuxiUzzNp1jpfBfx8yefKoO9D/RCR9dmwrnc7qjJqId1rEtQz+lSlU5cabX8daToMSx/7HayXOiaFw== + +"@hpke/core@1.9.0": + version "1.9.0" + resolved "https://registry.yarnpkg.com/@hpke/core/-/core-1.9.0.tgz#4eced0597787f51fdb53ad515c3d47a3abc9fe3a" + integrity sha512-pFxWl1nNJeQCSUFs7+GAblHvXBCjn9EPN65vdKlYQil2aURaRxfGMO6vBKGqm1YHTKwiAxJQNEI70PbSowMP9Q== + dependencies: + "@hpke/common" "^1.10.0" + "@humanfs/core@^0.19.2": version "0.19.2" resolved "https://registry.yarnpkg.com/@humanfs/core/-/core-0.19.2.tgz#a8272ca03b2acf492670222b2320b6c421bfde60" @@ -2615,6 +2627,13 @@ ts-api-utils@^2.5.0: resolved "https://registry.yarnpkg.com/ts-api-utils/-/ts-api-utils-2.5.0.tgz#4acd4a155e22734990a5ed1fe9e97f113bcb37c1" integrity sha512-OJ/ibxhPlqrMM0UiNHJ/0CKQkoKF243/AEmplt3qpRgkW8VG7IfOS41h7V8TjITqdByHzrjcS/2si+y4lIh8NA== +ts-mls@1.6.4: + version "1.6.4" + resolved "https://registry.yarnpkg.com/ts-mls/-/ts-mls-1.6.4.tgz#a2f35de3ed3d57a35dfa7247995e25bd97f828d6" + integrity sha512-BFb9qJ3V1+HIuifC+MtnlROEj3oc6NEzA7Ig1/LAmANDRhNay7UCnwkzijzHH9cav5wf1BoJw8KQ2mXiEhPy1Q== + dependencies: + "@hpke/core" "1.9.0" + ts-node@^10.9.2: version "10.9.2" resolved "https://registry.yarnpkg.com/ts-node/-/ts-node-10.9.2.tgz#70f021c9e185bccdca820e26dc413805c101c71f" From 1a07ffb9677019bcebc9ce8c9e79f659ef8840b1 Mon Sep 17 00:00:00 2001 From: Sivert Date: Fri, 25 Sep 2026 12:15:23 +0200 Subject: [PATCH 2/2] Name the MLS push recipients before emitting (GRYT-1500) The socket coverage check reads an emit's event name only when the first argument has no parentheses in it, so mls:message, mls:welcome and mls:devices:changed read as documented events the server didn't have. Co-Authored-By: Claude Opus 5.5 --- src/socket/handlers/mls.ts | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/socket/handlers/mls.ts b/src/socket/handlers/mls.ts index dc4be578..f8de0cc5 100644 --- a/src/socket/handlers/mls.ts +++ b/src/socket/handlers/mls.ts @@ -235,7 +235,8 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { const blockers = await blockersOfSender(entry.senderServerUserId); audience = memberIds.filter((id) => id === entry.senderServerUserId || !blockers.has(id)); } - emitTo(socketsOf(audience), "mls:message", entryView(conversationId, entry)); + const recipients = socketsOf(audience); + emitTo(recipients, "mls:message", entryView(conversationId, entry)); } /* Whoever shares a DM with them adds the new device the next time they send. */ @@ -244,7 +245,8 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { for (const c of await listConversationsForUser(serverUserId)) { for (const id of c.other_server_user_ids) people.add(id); } - emitTo(socketsOf(people), "mls:devices:changed", { serverUserId }); + const recipients = socketsOf(people); + emitTo(recipients, "mls:devices:changed", { serverUserId }); } return { @@ -508,7 +510,8 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { }; await fanOut(memberIds, entry, payload.conversationId); welcomes.forEach((w, i) => { - emitTo(socketsOf([w.serverUserId]), "mls:welcome", { + const recipients = socketsOf([w.serverUserId]); + emitTo(recipients, "mls:welcome", { welcomeId: result.welcomeIds[i], conversationId: payload.conversationId, groupId: group.groupId, @@ -602,7 +605,8 @@ export function registerMlsHandlers(ctx: HandlerContext): EventHandlerMap { if (wasEmpty) { for (const id of memberIds) { const view = (await directConversationViews(id)).find((v) => v.conversation_id === payload.conversationId); - if (view) emitTo(socketsOf([id]), "dm:opened", view); + const recipients = socketsOf([id]); + if (view) emitTo(recipients, "dm:opened", view); } } }