feat(federation): add peering handshake and admin peer management endpoints

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
2026-03-25 21:19:47 +01:00
parent 32a0c2e618
commit fbdec82172
2 changed files with 373 additions and 0 deletions
@@ -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<void> {
  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 () => {
@@ -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, string | string[] | undefined> }): 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<string, unknown>).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<string, number[]>();
+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<void> {
+  // ─── 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 });
+    },
+  );
+}