From fbdec821728fde237c5c161b62920e31e3f3168e Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 25 Mar 2026 21:19:47 +0100 Subject: [PATCH] feat(federation): add peering handshake and admin peer management endpoints MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add federation routes module with four endpoints: - POST /api/federation/peer/initiate — admin-only, starts peering handshake with a remote instance (validates origin, generates HMAC secret, sends server-to-server accept request with 10s timeout) - POST /api/federation/peer/accept — server-to-server first contact, rate-limited to 10 req/min per IP, creates/activates peer record - GET /api/federation/peers — admin-only, lists all peers (hmacSecret excluded) - DELETE /api/federation/peers/:id — admin-only, revokes peer and cleans outbox --- packages/server/src/index.ts | 2 + packages/server/src/routes/federation.ts | 371 +++++++++++++++++++++++ 2 files changed, 373 insertions(+) create mode 100644 packages/server/src/routes/federation.ts diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index a0289d47..5890eacd 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -25,6 +25,7 @@ import { exploreRoutes } from './routes/explore.js'; import { searchRoutes } from './routes/search.js'; import { adminRoutes } from './routes/admin.js'; import { gifRoutes } from './routes/gif.js'; +import { federationRoutes } from './routes/federation.js'; import { registerWebSocket } from './ws/handler.js'; import path from 'path'; @@ -95,6 +96,7 @@ async function main(): Promise { await app.register(searchRoutes); await app.register(adminRoutes); await app.register(gifRoutes); + await app.register(federationRoutes); await app.register(registerWebSocket); app.get('/api/health', async () => { diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts new file mode 100644 index 00000000..256ff817 --- /dev/null +++ b/packages/server/src/routes/federation.ts @@ -0,0 +1,371 @@ +import type { FastifyInstance } from 'fastify'; +import { randomBytes } from 'node:crypto'; +import { eq } from 'drizzle-orm'; +import { authenticate, requireAdmin } from '../utils/auth.js'; +import { generateHmacSecret } from '../utils/federationAuth.js'; +import { generateSnowflake } from '../utils/snowflake.js'; +import { getDb, schema } from '../db/index.js'; +import { config } from '../config.js'; + +/** Fields safe to expose to admin callers (everything except hmacSecret). */ +interface SanitizedPeer { + id: string; + origin: string; + instanceName: string | null; + status: string; + lastSeenAt: number | null; + lastFailureAt: number | null; + consecutiveFailures: number | null; + lastSyncedAt: number | null; + createdAt: number; +} + +function sanitizePeer(row: typeof schema.federationPeers.$inferSelect): SanitizedPeer { + return { + id: row.id, + origin: row.origin, + instanceName: row.instanceName, + status: row.status, + lastSeenAt: row.lastSeenAt, + lastFailureAt: row.lastFailureAt, + consecutiveFailures: row.consecutiveFailures, + lastSyncedAt: row.lastSyncedAt, + createdAt: row.createdAt, + }; +} + +/** + * Determine this instance's public origin. + * Prefer the explicit DOMAIN env var; fall back to the request Host header. + */ +function resolveLocalOrigin(request: { headers: Record }): string { + if (config.domain) { + return `https://${config.domain}`; + } + const hostHeader = request.headers.host; + const host = Array.isArray(hostHeader) ? hostHeader[0] : hostHeader; + if (!host) { + throw new Error('Cannot determine local origin: no DOMAIN configured and no Host header'); + } + const protocol = (request as Record).protocol === 'https' ? 'https' : 'http'; + return `${protocol}://${host}`; +} + +/** + * Validate that a string is a well-formed HTTP(S) URL origin. + * Returns the normalized origin (no trailing slash) or null if invalid. + */ +function validateOrigin(raw: string): string | null { + try { + const url = new URL(raw); + if (url.protocol !== 'http:' && url.protocol !== 'https:') return null; + // origin is scheme + host (+ port if non-default) — no trailing slash + return url.origin; + } catch { + return null; + } +} + +// ─── In-memory rate limiter for the accept endpoint ────────────────────────── +const acceptRateBuckets = new Map(); +const ACCEPT_RATE_WINDOW_MS = 60_000; +const ACCEPT_RATE_MAX = 10; + +function isAcceptRateLimited(ip: string): boolean { + const now = Date.now(); + let timestamps = acceptRateBuckets.get(ip); + if (!timestamps) { + timestamps = []; + acceptRateBuckets.set(ip, timestamps); + } + // Prune entries outside the window + const cutoff = now - ACCEPT_RATE_WINDOW_MS; + while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) { + timestamps.shift(); + } + if (timestamps.length >= ACCEPT_RATE_MAX) { + return true; + } + timestamps.push(now); + return false; +} + +// Periodically clean stale buckets to prevent unbounded memory growth +setInterval(() => { + const cutoff = Date.now() - ACCEPT_RATE_WINDOW_MS; + for (const [ip, timestamps] of acceptRateBuckets) { + // Remove expired entries + while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) { + timestamps.shift(); + } + if (timestamps.length === 0) { + acceptRateBuckets.delete(ip); + } + } +}, ACCEPT_RATE_WINDOW_MS).unref(); + +export async function federationRoutes(app: FastifyInstance): Promise { + // ─── POST /api/federation/peer/initiate ──────────────────────────────────── + // Admin-only: start a peering handshake with a remote instance. + app.post<{ Body: { remoteOrigin: string } }>( + '/api/federation/peer/initiate', + { preHandler: [authenticate, requireAdmin] }, + async (request, reply) => { + const { remoteOrigin: rawOrigin } = request.body ?? {}; + if (!rawOrigin || typeof rawOrigin !== 'string') { + return reply.code(400).send({ error: 'remoteOrigin is required', statusCode: 400 }); + } + + const remoteOrigin = validateOrigin(rawOrigin); + if (!remoteOrigin) { + return reply.code(400).send({ error: 'remoteOrigin must be a valid HTTP/HTTPS URL', statusCode: 400 }); + } + + const db = getDb(); + + // Check if a peer already exists for this origin + const existing = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, remoteOrigin)) + .get(); + + if (existing) { + if (existing.status === 'active') { + return reply.code(200).send({ peer: sanitizePeer(existing) }); + } + if (existing.status === 'pending') { + return reply.code(409).send({ + error: 'A peering handshake with this instance is already in progress', + statusCode: 409, + }); + } + // If revoked, allow re-initiation by removing the old record + if (existing.status === 'revoked') { + db.delete(schema.federationPeers).where(eq(schema.federationPeers.id, existing.id)).run(); + } + } + + let localOrigin: string; + try { + localOrigin = resolveLocalOrigin(request); + } catch { + return reply.code(500).send({ + error: 'Cannot determine local instance origin. Set the DOMAIN environment variable.', + statusCode: 500, + }); + } + + // Prevent self-peering + if (localOrigin === remoteOrigin) { + return reply.code(400).send({ error: 'Cannot peer with yourself', statusCode: 400 }); + } + + const hmacSecret = generateHmacSecret(); + const challenge = randomBytes(16).toString('hex'); + const peerId = generateSnowflake(); + const now = Date.now(); + + // Store peer as pending + db.insert(schema.federationPeers).values({ + id: peerId, + origin: remoteOrigin, + hmacSecret, + status: 'pending', + createdAt: now, + }).run(); + + // Initiate the server-to-server handshake + try { + const response = await fetch(`${remoteOrigin}/api/federation/peer/accept`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + sourceOrigin: localOrigin, + challenge, + hmacSecret, + }), + signal: AbortSignal.timeout(10_000), + }); + + if (!response.ok) { + let errorMessage = `Remote instance rejected peering (HTTP ${response.status})`; + try { + const body = await response.json() as { error?: string }; + if (body.error) { + errorMessage = body.error; + } + } catch { + // Ignore parse failures — use the default error message + } + + // Clean up the pending peer + db.delete(schema.federationPeers).where(eq(schema.federationPeers.id, peerId)).run(); + return reply.code(502).send({ error: errorMessage, statusCode: 502 }); + } + + // Remote accepted — activate the peer + db.update(schema.federationPeers) + .set({ status: 'active', lastSeenAt: Date.now() }) + .where(eq(schema.federationPeers.id, peerId)) + .run(); + + const peer = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.id, peerId)) + .get(); + + if (!peer) { + return reply.code(500).send({ error: 'Failed to read peer after activation', statusCode: 500 }); + } + + return reply.code(200).send({ peer: sanitizePeer(peer) }); + } catch (err: unknown) { + // Clean up the pending peer on network/timeout errors + db.delete(schema.federationPeers).where(eq(schema.federationPeers.id, peerId)).run(); + + const message = err instanceof Error ? err.message : 'Unknown error'; + if (err instanceof DOMException && err.name === 'TimeoutError') { + return reply.code(504).send({ + error: 'Remote instance did not respond within 10 seconds', + statusCode: 504, + }); + } + return reply.code(502).send({ + error: `Failed to reach remote instance: ${message}`, + statusCode: 502, + }); + } + }, + ); + + // ─── POST /api/federation/peer/accept ────────────────────────────────────── + // Server-to-server: accept a peering request from a remote instance. + // No JWT auth — this is first contact. Rate-limited by IP. + app.post<{ Body: { sourceOrigin: string; challenge: string; hmacSecret: string } }>( + '/api/federation/peer/accept', + async (request, reply) => { + const clientIp = request.ip; + if (isAcceptRateLimited(clientIp)) { + return reply.code(429).send({ + error: 'Too many peering requests — try again later', + statusCode: 429, + }); + } + + const { sourceOrigin: rawOrigin, challenge, hmacSecret } = request.body ?? {}; + + if (!rawOrigin || typeof rawOrigin !== 'string') { + return reply.code(400).send({ error: 'sourceOrigin is required', statusCode: 400 }); + } + if (!challenge || typeof challenge !== 'string') { + return reply.code(400).send({ error: 'challenge is required', statusCode: 400 }); + } + if (!hmacSecret || typeof hmacSecret !== 'string') { + return reply.code(400).send({ error: 'hmacSecret is required', statusCode: 400 }); + } + + const sourceOrigin = validateOrigin(rawOrigin); + if (!sourceOrigin) { + return reply.code(400).send({ error: 'sourceOrigin must be a valid HTTP/HTTPS URL', statusCode: 400 }); + } + + const db = getDb(); + + // Check if peer already exists + const existing = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.origin, sourceOrigin)) + .get(); + + if (existing) { + if (existing.status === 'active') { + // Idempotent — already peered + return reply.code(200).send({ accepted: true }); + } + if (existing.status === 'revoked') { + return reply.code(403).send({ + error: 'Peering with this instance has been revoked', + statusCode: 403, + }); + } + // Pending — update with new secret and activate + db.update(schema.federationPeers) + .set({ + hmacSecret, + status: 'active', + lastSeenAt: Date.now(), + }) + .where(eq(schema.federationPeers.id, existing.id)) + .run(); + + return reply.code(200).send({ accepted: true }); + } + + // New peer — create and activate + const peerId = generateSnowflake(); + db.insert(schema.federationPeers).values({ + id: peerId, + origin: sourceOrigin, + hmacSecret, + status: 'active', + lastSeenAt: Date.now(), + createdAt: Date.now(), + }).run(); + + return reply.code(200).send({ accepted: true }); + }, + ); + + // ─── GET /api/federation/peers ───────────────────────────────────────────── + // Admin-only: list all federation peers (hmacSecret excluded). + app.get( + '/api/federation/peers', + { preHandler: [authenticate, requireAdmin] }, + async (_request, reply) => { + const db = getDb(); + const peers = db + .select() + .from(schema.federationPeers) + .all(); + + return reply.code(200).send({ peers: peers.map(sanitizePeer) }); + }, + ); + + // ─── DELETE /api/federation/peers/:id ────────────────────────────────────── + // Admin-only: revoke a federation peer and clean up its outbox. + app.delete<{ Params: { id: string } }>( + '/api/federation/peers/:id', + { preHandler: [authenticate, requireAdmin] }, + async (request, reply) => { + const { id } = request.params; + const db = getDb(); + + const peer = db + .select() + .from(schema.federationPeers) + .where(eq(schema.federationPeers.id, id)) + .get(); + + if (!peer) { + return reply.code(404).send({ error: 'Peer not found', statusCode: 404 }); + } + + // Revoke the peer + db.update(schema.federationPeers) + .set({ status: 'revoked' }) + .where(eq(schema.federationPeers.id, id)) + .run(); + + // Delete all outbox entries for this peer + db.delete(schema.federationOutbox) + .where(eq(schema.federationOutbox.peerId, id)) + .run(); + + return reply.code(200).send({ success: true }); + }, + ); +}