diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index 256ff817..af7116b3 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -1,11 +1,17 @@ import type { FastifyInstance } from 'fastify'; import { randomBytes } from 'node:crypto'; -import { eq } from 'drizzle-orm'; +import { eq, and, or, isNull } from 'drizzle-orm'; import { authenticate, requireAdmin } from '../utils/auth.js'; -import { generateHmacSecret } from '../utils/federationAuth.js'; +import { generateHmacSecret, parseFederationHeaders, verifySignature } from '../utils/federationAuth.js'; import { generateSnowflake } from '../utils/snowflake.js'; import { getDb, schema } from '../db/index.js'; import { config } from '../config.js'; +import { connectionManager } from '../ws/handler.js'; +import { sanitizeUser } from '../utils/sanitize.js'; +import { deleteAttachmentFiles } from '../utils/fileCleanup.js'; +import { canonicalDmPairId } from '../utils/federationOutbox.js'; +import { broadcastDmMessage } from './dm.js'; +import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, DmMessageWithUser } from '@backspace/shared'; /** Fields safe to expose to admin callers (everything except hmacSecret). */ interface SanitizedPeer { @@ -368,4 +374,762 @@ export async function federationRoutes(app: FastifyInstance): Promise { return reply.code(200).send({ success: true }); }, ); + + // ─── POST /api/federation/relay ──────────────────────────────────────────── + // Server-to-server: receive relayed DM events from a peer instance. + // Authenticated via HMAC-SHA256 signature, NOT JWT. + app.post<{ Body: FederationRelayRequest }>( + '/api/federation/relay', + { bodyLimit: 10 * 1024 * 1024 }, + async (request, reply) => { + const db = getDb(); + + // 1. Verify HMAC signature + const fedHeaders = parseFederationHeaders(request.headers as Record); + if (!fedHeaders) { + return reply.code(401).send({ error: 'Missing or malformed federation headers', statusCode: 401 }); + } + + const peer = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, fedHeaders.origin)) + .get(); + + if (!peer || peer.status !== 'active') { + return reply.code(403).send({ error: 'Unknown or inactive peer', statusCode: 403 }); + } + + // Serialize body back to JSON for HMAC verification (we control both sides) + const bodyString = JSON.stringify(request.body); + if (!verifySignature(bodyString, fedHeaders.signature, peer.hmacSecret, fedHeaders.timestamp)) { + return reply.code(401).send({ error: 'Invalid signature', statusCode: 401 }); + } + + // 2. Validate request body shape + const body = request.body; + if (!body || body.version !== 1 || !Array.isArray(body.events)) { + return reply.code(400).send({ error: 'Invalid relay request format', statusCode: 400 }); + } + + if (body.events.length > 50) { + return reply.code(400).send({ error: 'Maximum 50 events per batch', statusCode: 400 }); + } + + const sourceInstance = body.sourceInstance; + if (!sourceInstance || typeof sourceInstance !== 'string') { + return reply.code(400).send({ error: 'sourceInstance is required', statusCode: 400 }); + } + + // 3. Process each event + const accepted: string[] = []; + const rejected: Array<{ messageId: string; reason: string }> = []; + + for (const event of body.events) { + try { + switch (event.eventType) { + case 'create': + processCreateEvent(event, sourceInstance, peer.origin, db, accepted, rejected); + break; + case 'update': + processUpdateEvent(event, sourceInstance, db, accepted, rejected); + break; + case 'delete': + processDeleteEvent(event, sourceInstance, db, accepted, rejected); + break; + case 'reaction_add': + processReactionAddEvent(event, sourceInstance, db, accepted, rejected); + break; + case 'reaction_remove': + processReactionRemoveEvent(event, sourceInstance, db, accepted, rejected); + break; + default: + rejected.push({ messageId: event.messageId, reason: 'unknown_event_type' }); + break; + } + } catch (err) { + const errMsg = err instanceof Error ? err.message : 'unknown_error'; + console.error(`[federation-relay] Error processing event ${event.messageId}:`, errMsg); + rejected.push({ messageId: event.messageId, reason: 'processing_error' }); + } + } + + // 4. Update peer status + db.update(schema.federationPeers) + .set({ + lastSeenAt: Date.now(), + consecutiveFailures: 0, + }) + .where(eq(schema.federationPeers.id, peer.id)) + .run(); + + // 5. Return response with max upload size info + const settings = db + .select({ maxUploadSizeBytes: schema.instanceSettings.maxUploadSizeBytes }) + .from(schema.instanceSettings) + .where(eq(schema.instanceSettings.id, 1)) + .get(); + + const response: FederationRelayResponse = { + accepted, + rejected, + maxUploadSize: settings?.maxUploadSizeBytes ?? config.maxUploadSize, + }; + + return reply.code(200).send(response); + }, + ); +} + +// ─── Relay Event Processors ────────────────────────────────────────────────── + +/** + * Resolve a home user ID to a local user. + * Matches users where home_user_id = homeUserId, or where + * the user's own id equals homeUserId and they have no home_instance set (local user). + */ +function resolveLocalUser( + homeUserId: string, + db: ReturnType, +): typeof schema.users.$inferSelect | undefined { + return db + .select() + .from(schema.users) + .where( + or( + eq(schema.users.homeUserId, homeUserId), + and(eq(schema.users.id, homeUserId), isNull(schema.users.homeInstance)), + ), + ) + .get(); +} + +/** + * Find or create a local DM channel for a federated 1-on-1 pair. + * Uses canonical_pair_id for deterministic cross-instance lookup. + */ +function findOrCreateDmChannel( + canonicalPairId: string, + localUserIdA: string, + localUserIdB: string, + db: ReturnType, +): string { + // Try to find existing channel by canonical pair ID + const existing = db + .select() + .from(schema.dmChannels) + .where(eq(schema.dmChannels.canonicalPairId, canonicalPairId)) + .get(); + + if (existing) { + // Ensure both users are members (they might have been removed) + for (const userId of [localUserIdA, localUserIdB]) { + const member = db + .select() + .from(schema.dmMembers) + .where( + and( + eq(schema.dmMembers.dmChannelId, existing.id), + eq(schema.dmMembers.userId, userId), + ), + ) + .get(); + + if (!member) { + db.insert(schema.dmMembers) + .values({ + dmChannelId: existing.id, + userId, + closed: 0, + }) + .run(); + } + } + return existing.id; + } + + // Create new DM channel with canonical pair ID + const channelId = generateSnowflake(); + const now = Date.now(); + + db.insert(schema.dmChannels) + .values({ + id: channelId, + canonicalPairId, + createdAt: now, + }) + .run(); + + for (const userId of [localUserIdA, localUserIdB]) { + db.insert(schema.dmMembers) + .values({ + dmChannelId: channelId, + userId, + closed: 0, + }) + .run(); + } + + return channelId; +} + +/** + * Build a DmMessageWithUser payload for WebSocket broadcasting. + */ +function buildDmMessagePayload( + messageRow: { + id: string; + dmChannelId: string; + userId: string; + content: string | null; + replyToId: string | null; + editedAt: number | null; + createdAt: number; + }, + userRow: typeof schema.users.$inferSelect, +): DmMessageWithUser { + return { + id: messageRow.id, + dmChannelId: messageRow.dmChannelId, + channelId: messageRow.dmChannelId, + userId: messageRow.userId, + content: messageRow.content, + replyToId: messageRow.replyToId, + editedAt: messageRow.editedAt, + createdAt: messageRow.createdAt, + user: sanitizeUser(userRow), + attachments: [], + embeds: [], + reactions: [], + }; +} + +/** + * Validate that a URL's hostname matches the peer origin's hostname (SSRF protection). + */ +function isUrlFromPeer(sourceUrl: string, peerOrigin: string): boolean { + try { + const sourceHost = new URL(sourceUrl).hostname; + const peerHost = new URL(peerOrigin).hostname; + return sourceHost === peerHost; + } catch { + return false; + } +} + +function processCreateEvent( + event: FederationRelayEvent, + sourceInstance: string, + peerOrigin: string, + db: ReturnType, + accepted: string[], + rejected: Array<{ messageId: string; reason: string }>, +): void { + if (!event.message) { + rejected.push({ messageId: event.messageId, reason: 'missing_message_payload' }); + return; + } + + // Resolve the message author to a local user + const authorUser = resolveLocalUser(event.message.homeUserId, db); + if (!authorUser) { + rejected.push({ messageId: event.messageId, reason: 'user_not_found' }); + return; + } + + // Dedup: check for existing message with same source + const existingMsg = db + .select() + .from(schema.dmMessages) + .where( + and( + eq(schema.dmMessages.sourceInstance, sourceInstance), + eq(schema.dmMessages.sourceMessageId, event.messageId), + ), + ) + .get(); + + if (existingMsg) { + rejected.push({ messageId: event.messageId, reason: 'duplicate' }); + return; + } + + // Resolve the DM recipient. Federated DMs are 1-on-1: one side is the author, + // the other is a local user on this instance. We match via canonical_pair_id. + const authorHomeUserId = event.message.homeUserId; + + // First, search existing DM channels where the author is already a member + const authorMemberships = db + .select({ dmChannelId: schema.dmMembers.dmChannelId }) + .from(schema.dmMembers) + .where(eq(schema.dmMembers.userId, authorUser.id)) + .all(); + + let localDmChannelId: string | null = null; + + // Check each of the author's DM channels to find the matching one + for (const membership of authorMemberships) { + const channelMembers = db + .select() + .from(schema.dmMembers) + .where(eq(schema.dmMembers.dmChannelId, membership.dmChannelId)) + .all(); + + // For 1-on-1 DMs, there should be exactly 2 members + if (channelMembers.length === 2) { + const otherMember = channelMembers.find(m => m.userId !== authorUser.id); + if (otherMember) { + const otherUser = db + .select() + .from(schema.users) + .where(eq(schema.users.id, otherMember.userId)) + .get(); + + if (otherUser) { + const otherHomeUserId = otherUser.homeUserId || otherUser.id; + const pairId = canonicalDmPairId(authorHomeUserId, otherHomeUserId); + const channel = db + .select() + .from(schema.dmChannels) + .where(eq(schema.dmChannels.id, membership.dmChannelId)) + .get(); + + if (channel?.canonicalPairId === pairId) { + localDmChannelId = membership.dmChannelId; + break; + } + } + } + } + } + + // If no existing channel found, search the author's friends for the recipient. + // On cold start (first federated DM), we use the friends list as a hint to + // find the local user and create the DM channel. + if (!localDmChannelId) { + const friendRows = db + .select() + .from(schema.friends) + .where( + or( + eq(schema.friends.userId, authorUser.id), + eq(schema.friends.friendId, authorUser.id), + ), + ) + .all(); + + const friendIds = friendRows.map(f => + f.userId === authorUser.id ? f.friendId : f.userId, + ); + + for (const friendId of friendIds) { + const friendUser = db + .select() + .from(schema.users) + .where(eq(schema.users.id, friendId)) + .get(); + + if (friendUser) { + const friendHomeUserId = friendUser.homeUserId || friendUser.id; + const pairId = canonicalDmPairId(authorHomeUserId, friendHomeUserId); + + // Check if a channel already exists with this pair ID + const existingChannel = db + .select() + .from(schema.dmChannels) + .where(eq(schema.dmChannels.canonicalPairId, pairId)) + .get(); + + if (existingChannel) { + localDmChannelId = existingChannel.id; + break; + } + } + } + + // If still no channel with a matching canonical pair ID, create one. + // The recipient must be a local (non-federated) user who is friends with the author. + if (!localDmChannelId && friendIds.length > 0) { + for (const friendId of friendIds) { + const friendUser = db + .select() + .from(schema.users) + .where(eq(schema.users.id, friendId)) + .get(); + + if (friendUser && !friendUser.homeInstance) { + // This is a local user — they're a candidate recipient + const friendHomeUserId = friendUser.homeUserId || friendUser.id; + const pairId = canonicalDmPairId(authorHomeUserId, friendHomeUserId); + + localDmChannelId = findOrCreateDmChannel(pairId, authorUser.id, friendId, db); + break; + } + } + } + } + + if (!localDmChannelId) { + rejected.push({ messageId: event.messageId, reason: 'recipient_not_found' }); + return; + } + + // Insert the message + const localMessageId = generateSnowflake(); + db.insert(schema.dmMessages) + .values({ + id: localMessageId, + dmChannelId: localDmChannelId, + userId: authorUser.id, + content: event.message.content, + replyToId: null, + createdAt: event.message.createdAt, + editedAt: null, + sourceInstance, + sourceMessageId: event.messageId, + encryptionVersion: 0, + }) + .run(); + + // Queue attachment downloads (SSRF-validated) + if (event.message.attachments && event.message.attachments.length > 0) { + const now = Date.now(); + for (const attachment of event.message.attachments) { + if (!isUrlFromPeer(attachment.sourceUrl, peerOrigin)) { + console.warn( + `[federation-relay] Rejecting attachment URL ${attachment.sourceUrl} — hostname does not match peer ${peerOrigin}`, + ); + continue; + } + + db.insert(schema.federationFileQueue) + .values({ + id: generateSnowflake(), + peerOrigin, + dmMessageId: localMessageId, + sourceUrl: attachment.sourceUrl, + originalName: attachment.originalName, + mimetype: attachment.mimetype, + size: attachment.size, + status: 'pending', + nextRetryAt: now, + expiresAt: now + 30 * 86_400_000, + createdAt: now, + }) + .run(); + } + } + + // Broadcast to local WebSocket clients + const messagePayload = buildDmMessagePayload( + { + id: localMessageId, + dmChannelId: localDmChannelId, + userId: authorUser.id, + content: event.message.content, + replyToId: null, + editedAt: null, + createdAt: event.message.createdAt, + }, + authorUser, + ); + broadcastDmMessage(localDmChannelId, messagePayload); + + accepted.push(event.messageId); +} + +function processUpdateEvent( + event: FederationRelayEvent, + sourceInstance: string, + db: ReturnType, + accepted: string[], + rejected: Array<{ messageId: string; reason: string }>, +): void { + const localMsg = db + .select() + .from(schema.dmMessages) + .where( + and( + eq(schema.dmMessages.sourceInstance, sourceInstance), + eq(schema.dmMessages.sourceMessageId, event.messageId), + ), + ) + .get(); + + if (!localMsg) { + rejected.push({ messageId: event.messageId, reason: 'unknown_message' }); + return; + } + + const content = event.message?.content ?? null; + const editedAt = event.message?.editedAt ?? Date.now(); + + db.update(schema.dmMessages) + .set({ content, editedAt }) + .where(eq(schema.dmMessages.id, localMsg.id)) + .run(); + + // Broadcast update to local clients + const authorUser = db + .select() + .from(schema.users) + .where(eq(schema.users.id, localMsg.userId)) + .get(); + + if (authorUser) { + const updatedPayload = buildDmMessagePayload( + { + id: localMsg.id, + dmChannelId: localMsg.dmChannelId, + userId: localMsg.userId, + content, + replyToId: localMsg.replyToId, + editedAt, + createdAt: localMsg.createdAt, + }, + authorUser, + ); + + // Re-fetch reactions and attachments for the complete payload + const reactions = db + .select() + .from(schema.dmReactions) + .where(eq(schema.dmReactions.dmMessageId, localMsg.id)) + .all(); + + const attachments = db + .select() + .from(schema.attachments) + .where(eq(schema.attachments.dmMessageId, localMsg.id)) + .all(); + + updatedPayload.reactions = reactions.map(r => ({ + id: r.id, + messageId: r.dmMessageId, + userId: r.userId, + emoji: r.emoji, + createdAt: r.createdAt, + })); + + updatedPayload.attachments = attachments.map(a => ({ + id: a.id, + messageId: a.dmMessageId ?? a.messageId ?? '', + filename: a.filename, + originalName: a.originalName, + mimetype: a.mimetype, + size: a.size, + thumbnailFilename: a.thumbnailFilename, + width: a.width, + height: a.height, + duration: a.duration, + createdAt: a.createdAt, + })); + + connectionManager.sendToDmMembers(localMsg.dmChannelId, { + type: 'dm_message_updated', + message: updatedPayload, + }); + } + + accepted.push(event.messageId); +} + +function processDeleteEvent( + event: FederationRelayEvent, + sourceInstance: string, + db: ReturnType, + accepted: string[], + rejected: Array<{ messageId: string; reason: string }>, +): void { + const localMsg = db + .select() + .from(schema.dmMessages) + .where( + and( + eq(schema.dmMessages.sourceInstance, sourceInstance), + eq(schema.dmMessages.sourceMessageId, event.messageId), + ), + ) + .get(); + + if (!localMsg) { + rejected.push({ messageId: event.messageId, reason: 'unknown_message' }); + return; + } + + // Collect attachment filenames before deletion for disk cleanup + const attachmentRows = db + .select({ filename: schema.attachments.filename }) + .from(schema.attachments) + .where(eq(schema.attachments.dmMessageId, localMsg.id)) + .all(); + + // Delete attachments, reactions, and message atomically + db.transaction((tx) => { + tx.delete(schema.attachments) + .where(eq(schema.attachments.dmMessageId, localMsg.id)) + .run(); + tx.delete(schema.dmReactions) + .where(eq(schema.dmReactions.dmMessageId, localMsg.id)) + .run(); + tx.delete(schema.dmMessages) + .where(eq(schema.dmMessages.id, localMsg.id)) + .run(); + }); + + // Clean up files from disk + deleteAttachmentFiles(attachmentRows); + + // Broadcast deletion to local clients + connectionManager.sendToDmMembers(localMsg.dmChannelId, { + type: 'dm_message_deleted', + messageId: localMsg.id, + dmChannelId: localMsg.dmChannelId, + }); + + accepted.push(event.messageId); +} + +function processReactionAddEvent( + event: FederationRelayEvent, + sourceInstance: string, + db: ReturnType, + accepted: string[], + rejected: Array<{ messageId: string; reason: string }>, +): void { + if (!event.reaction) { + rejected.push({ messageId: event.messageId, reason: 'missing_reaction_payload' }); + return; + } + + // Find the local message corresponding to the source message + const localMsg = db + .select() + .from(schema.dmMessages) + .where( + and( + eq(schema.dmMessages.sourceInstance, sourceInstance), + eq(schema.dmMessages.sourceMessageId, event.messageId), + ), + ) + .get(); + + if (!localMsg) { + rejected.push({ messageId: event.messageId, reason: 'unknown_message' }); + return; + } + + // Resolve the reacting user + const reactingUser = resolveLocalUser(event.reaction.homeUserId, db); + if (!reactingUser) { + rejected.push({ messageId: event.messageId, reason: 'user_not_found' }); + return; + } + + // Dedup: check if this user already reacted with this emoji + const existingReaction = db + .select() + .from(schema.dmReactions) + .where( + and( + eq(schema.dmReactions.dmMessageId, localMsg.id), + eq(schema.dmReactions.userId, reactingUser.id), + eq(schema.dmReactions.emoji, event.reaction.emoji), + ), + ) + .get(); + + if (existingReaction) { + // Already exists — treat as accepted (idempotent) + accepted.push(event.messageId); + return; + } + + const reactionId = generateSnowflake(); + const now = event.reaction.createdAt || Date.now(); + + db.insert(schema.dmReactions) + .values({ + id: reactionId, + dmMessageId: localMsg.id, + userId: reactingUser.id, + emoji: event.reaction.emoji, + createdAt: now, + }) + .run(); + + // Broadcast to local clients + connectionManager.sendToDmMembers(localMsg.dmChannelId, { + type: 'reaction_added', + messageId: localMsg.id, + reaction: { + id: reactionId, + messageId: localMsg.id, + userId: reactingUser.id, + emoji: event.reaction.emoji, + createdAt: now, + user: sanitizeUser(reactingUser), + }, + }); + + accepted.push(event.messageId); +} + +function processReactionRemoveEvent( + event: FederationRelayEvent, + sourceInstance: string, + db: ReturnType, + accepted: string[], + rejected: Array<{ messageId: string; reason: string }>, +): void { + if (!event.reaction) { + rejected.push({ messageId: event.messageId, reason: 'missing_reaction_payload' }); + return; + } + + // Find the local message + const localMsg = db + .select() + .from(schema.dmMessages) + .where( + and( + eq(schema.dmMessages.sourceInstance, sourceInstance), + eq(schema.dmMessages.sourceMessageId, event.messageId), + ), + ) + .get(); + + if (!localMsg) { + rejected.push({ messageId: event.messageId, reason: 'unknown_message' }); + return; + } + + // Resolve the reacting user + const reactingUser = resolveLocalUser(event.reaction.homeUserId, db); + if (!reactingUser) { + rejected.push({ messageId: event.messageId, reason: 'user_not_found' }); + return; + } + + const result = db + .delete(schema.dmReactions) + .where( + and( + eq(schema.dmReactions.dmMessageId, localMsg.id), + eq(schema.dmReactions.userId, reactingUser.id), + eq(schema.dmReactions.emoji, event.reaction.emoji), + ), + ) + .run(); + + if (result.changes > 0) { + connectionManager.sendToDmMembers(localMsg.dmChannelId, { + type: 'reaction_removed', + messageId: localMsg.id, + userId: reactingUser.id, + emoji: event.reaction.emoji, + }); + } + + accepted.push(event.messageId); }