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..f8de0cc5 --- /dev/null +++ b/src/socket/handlers/mls.ts @@ -0,0 +1,713 @@ +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)); + } + 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. */ + 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); + } + const recipients = socketsOf(people); + emitTo(recipients, "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) => { + const recipients = socketsOf([w.serverUserId]); + emitTo(recipients, "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); + const recipients = socketsOf([id]); + if (view) emitTo(recipients, "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"