refactor(server): dedupe federation rate limiters and response signing (#10)
Phase C cleanup follow-up to the routes/federation split (#9). Behavior- preserving; full server suite (790 tests) green. A) rateLimits.ts: the four near-identical sliding-window limiters (accept/relay/lookup/ensure) and their duplicated prune loops collapse into one createLimiter(windowMs, max) factory. Per-call and periodic- sweep semantics are preserved exactly, including that lookup buckets are pruned per-call but never swept (unchanged from before). 177 -> 101 lines. B) Extract sendSignedJson(reply, payload, hmacSecret) — the single definition of how this instance signs an S2S JSON response — and use it in the /epoch and /verify-attach-proof|reattach handlers, replacing two copies of the build-headers-and-send boilerplate.
This commit is contained in:
TheZwiss
@@ -3,7 +3,8 @@ import { config } from '../../../config.js';
|
||||
import { getDb, getRawDb, schema } from '../../../db/index.js';
|
||||
import { authenticate } from '../../../utils/auth.js';
|
||||
import { fetchHomeProfileByHomeId, verifyAttachProofWithPeer } from '../../../utils/federationAttach.js';
|
||||
import { buildFederationHeaders, getOurOrigin, parseFederationHeaders, verifyPeerSignature } from '../../../utils/federationAuth.js';
|
||||
import { parseFederationHeaders, verifyPeerSignature } from '../../../utils/federationAuth.js';
|
||||
import { sendSignedJson } from './signedResponse.js';
|
||||
import { sanitizeUser } from '../../../utils/sanitize.js';
|
||||
import { collectProfileBroadcastTargetIds } from '../../../utils/userDeletion.js';
|
||||
import { connectionManager } from '../../../ws/handler.js';
|
||||
@@ -72,17 +73,8 @@ export function registerAttachRoutes(app: FastifyInstance): void {
|
||||
|
||||
// 2. Sign every downstream response with the peer's shared secret so the
|
||||
// caller can trust the identity (or the fail-closed verdict) it carries.
|
||||
const sendSigned = (payload: { valid: false } | { valid: true; homeUserId: string; username: string }): FastifyReply => {
|
||||
const responseBody = JSON.stringify(payload);
|
||||
const sigHeaders = buildFederationHeaders(responseBody, peer.hmacSecret, getOurOrigin());
|
||||
reply.headers({
|
||||
'X-Federation-Signature': sigHeaders['X-Federation-Signature'],
|
||||
'X-Federation-Timestamp': sigHeaders['X-Federation-Timestamp'],
|
||||
'X-Federation-Nonce': sigHeaders['X-Federation-Nonce'],
|
||||
'Content-Type': 'application/json',
|
||||
});
|
||||
return reply.code(200).send(responseBody);
|
||||
};
|
||||
const sendSigned = (payload: { valid: false } | { valid: true; homeUserId: string; username: string }): FastifyReply =>
|
||||
sendSignedJson(reply, payload, peer.hmacSecret);
|
||||
|
||||
// 3. Validate the token shape (64 hex chars, as minted by attach-proof).
|
||||
const rawToken = (request.body as { token?: unknown } | null)?.token;
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import path from 'node:path';
|
||||
import { config } from '../../../config.js';
|
||||
import { getDb, getRawDb, schema } from '../../../db/index.js';
|
||||
import { buildFederationHeaders, getOurOrigin, parseFederationHeaders, verifyPeerSignature } from '../../../utils/federationAuth.js';
|
||||
import { getOurOrigin, parseFederationHeaders, verifyPeerSignature } from '../../../utils/federationAuth.js';
|
||||
import { sendSignedJson } from './signedResponse.js';
|
||||
import { getInstanceId } from '../../../utils/federationEpoch.js';
|
||||
import { getDmParticipants } from '../../../utils/federationOutbox.js';
|
||||
import { deleteAttachmentFiles } from '../../../utils/fileCleanup.js';
|
||||
@@ -285,15 +286,7 @@ export function registerRelayRoutes(app: FastifyInstance): void {
|
||||
}
|
||||
|
||||
// 4. Sign the response body with the peer's shared secret and return it.
|
||||
const responseBody = JSON.stringify({ instanceId: getInstanceId() });
|
||||
const sigHeaders = buildFederationHeaders(responseBody, peer.hmacSecret, getOurOrigin());
|
||||
reply.headers({
|
||||
'X-Federation-Signature': sigHeaders['X-Federation-Signature'],
|
||||
'X-Federation-Timestamp': sigHeaders['X-Federation-Timestamp'],
|
||||
'X-Federation-Nonce': sigHeaders['X-Federation-Nonce'],
|
||||
'Content-Type': 'application/json',
|
||||
});
|
||||
return reply.code(200).send(responseBody);
|
||||
return sendSignedJson(reply, { instanceId: getInstanceId() }, peer.hmacSecret);
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -0,0 +1,24 @@
|
||||
import type { FastifyReply } from 'fastify';
|
||||
import { buildFederationHeaders, getOurOrigin } from '../../../utils/federationAuth.js';
|
||||
|
||||
/**
|
||||
* Serialize `payload` as JSON and send it as a `200` response signed with the
|
||||
* peer's shared HMAC secret, so the receiving instance can verify authenticity
|
||||
* (or trust a fail-closed verdict) of the body it carries.
|
||||
*
|
||||
* This is the single definition of how this instance signs S2S responses: the
|
||||
* body is stringified once and the signature is computed over those exact bytes,
|
||||
* which are the bytes sent (Content-Type is set explicitly so Fastify does not
|
||||
* re-serialize and desync the signature).
|
||||
*/
|
||||
export function sendSignedJson(reply: FastifyReply, payload: unknown, hmacSecret: string): FastifyReply {
|
||||
const responseBody = JSON.stringify(payload);
|
||||
const sigHeaders = buildFederationHeaders(responseBody, hmacSecret, getOurOrigin());
|
||||
reply.headers({
|
||||
'X-Federation-Signature': sigHeaders['X-Federation-Signature'],
|
||||
'X-Federation-Timestamp': sigHeaders['X-Federation-Timestamp'],
|
||||
'X-Federation-Nonce': sigHeaders['X-Federation-Nonce'],
|
||||
'Content-Type': 'application/json',
|
||||
});
|
||||
return reply.code(200).send(responseBody);
|
||||
}
|
||||
@@ -1,138 +1,74 @@
|
||||
// ─── In-memory sliding-window rate limiters (federation S2S endpoints) ───────
|
||||
//
|
||||
// Each limiter keeps a per-key ring of request timestamps inside a fixed window.
|
||||
// `limited(key)` prunes that key's expired entries, then returns true (without
|
||||
// recording a hit) once the key is at capacity. `sweep()` prunes every key and
|
||||
// drops emptied buckets to bound memory; it runs on a timer, not per request.
|
||||
|
||||
|
||||
// ─── In-memory rate limiter for the accept endpoint ──────────────────────────
|
||||
export const acceptRateBuckets = new Map<string, number[]>();
|
||||
|
||||
export const ACCEPT_RATE_WINDOW_MS = 60_000;
|
||||
|
||||
export const ACCEPT_RATE_MAX = 10;
|
||||
|
||||
|
||||
export 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;
|
||||
interface SlidingWindowLimiter {
|
||||
/** True if `key` is already at capacity for the current window; otherwise records the hit and returns false. */
|
||||
limited(key: string): boolean;
|
||||
/** Prune expired timestamps across all keys and drop now-empty buckets. */
|
||||
sweep(): void;
|
||||
/** Underlying buckets — exposed only so tests can reset state. */
|
||||
readonly buckets: Map<string, number[]>;
|
||||
}
|
||||
|
||||
|
||||
// ─── In-memory rate limiter for the relay endpoint (per-peer) ────────────────
|
||||
export const relayRateBuckets = new Map<string, number[]>();
|
||||
|
||||
export const RELAY_RATE_WINDOW_MS = 60_000;
|
||||
|
||||
export const RELAY_RATE_MAX = 90;
|
||||
|
||||
|
||||
export function isRelayRateLimited(peerOrigin: string): boolean {
|
||||
const now = Date.now();
|
||||
let timestamps = relayRateBuckets.get(peerOrigin);
|
||||
if (!timestamps) {
|
||||
timestamps = [];
|
||||
relayRateBuckets.set(peerOrigin, timestamps);
|
||||
}
|
||||
const cutoff = now - RELAY_RATE_WINDOW_MS;
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length >= RELAY_RATE_MAX) {
|
||||
return true;
|
||||
}
|
||||
timestamps.push(now);
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
// ─── In-memory rate limiter for the user-lookup endpoint (per-peer) ──────────
|
||||
export const lookupRateBuckets = new Map<string, number[]>();
|
||||
|
||||
export const LOOKUP_RATE_WINDOW_MS = 60_000;
|
||||
|
||||
export const LOOKUP_RATE_MAX = 60;
|
||||
|
||||
|
||||
export function isLookupRateLimited(peerOrigin: string): boolean {
|
||||
const now = Date.now();
|
||||
let timestamps = lookupRateBuckets.get(peerOrigin);
|
||||
if (!timestamps) {
|
||||
timestamps = [];
|
||||
lookupRateBuckets.set(peerOrigin, timestamps);
|
||||
}
|
||||
const cutoff = now - LOOKUP_RATE_WINDOW_MS;
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length >= LOOKUP_RATE_MAX) return true;
|
||||
timestamps.push(now);
|
||||
return false;
|
||||
}
|
||||
|
||||
|
||||
// Test-only export — used by federation.userLookup.test.ts to reset between cases.
|
||||
export function _resetLookupRateBuckets(): void {
|
||||
lookupRateBuckets.clear();
|
||||
}
|
||||
|
||||
|
||||
// ─── In-memory rate limiter for the ensure endpoint (per-user) ─────────────
|
||||
export const ensureRateBuckets = new Map<string, number[]>();
|
||||
|
||||
export const ENSURE_RATE_WINDOW_MS = 15 * 60_000; // 15 minutes
|
||||
|
||||
export const ENSURE_RATE_MAX = 3;
|
||||
|
||||
|
||||
export function isEnsureRateLimited(userId: string): boolean {
|
||||
const now = Date.now();
|
||||
let timestamps = ensureRateBuckets.get(userId);
|
||||
if (!timestamps) {
|
||||
timestamps = [];
|
||||
ensureRateBuckets.set(userId, timestamps);
|
||||
}
|
||||
const cutoff = now - ENSURE_RATE_WINDOW_MS;
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length >= ENSURE_RATE_MAX) {
|
||||
return true;
|
||||
}
|
||||
timestamps.push(now);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Clean up stale ensure rate limit buckets every 15 minutes
|
||||
setInterval(() => {
|
||||
const cutoff = Date.now() - ENSURE_RATE_WINDOW_MS;
|
||||
for (const [userId, timestamps] of ensureRateBuckets) {
|
||||
function createLimiter(windowMs: number, max: number): SlidingWindowLimiter {
|
||||
const buckets = new Map<string, number[]>();
|
||||
const prune = (timestamps: number[], cutoff: number): void => {
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length === 0) {
|
||||
ensureRateBuckets.delete(userId);
|
||||
}
|
||||
}
|
||||
}, ENSURE_RATE_WINDOW_MS).unref();
|
||||
};
|
||||
return {
|
||||
buckets,
|
||||
limited(key: string): boolean {
|
||||
const now = Date.now();
|
||||
let timestamps = buckets.get(key);
|
||||
if (!timestamps) {
|
||||
timestamps = [];
|
||||
buckets.set(key, timestamps);
|
||||
}
|
||||
prune(timestamps, now - windowMs);
|
||||
if (timestamps.length >= max) return true;
|
||||
timestamps.push(now);
|
||||
return false;
|
||||
},
|
||||
sweep(): void {
|
||||
const cutoff = Date.now() - windowMs;
|
||||
for (const [key, timestamps] of buckets) {
|
||||
prune(timestamps, cutoff);
|
||||
if (timestamps.length === 0) buckets.delete(key);
|
||||
}
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const RATE_WINDOW_MS = 60_000;
|
||||
const ENSURE_WINDOW_MS = 15 * 60_000;
|
||||
|
||||
// ─── In-memory nonce store for replay protection (per-peer) ──────────────────
|
||||
// accept: per source IP · relay & user-lookup: per peer origin · ensure: per user
|
||||
const acceptLimiter = createLimiter(RATE_WINDOW_MS, 10);
|
||||
const relayLimiter = createLimiter(RATE_WINDOW_MS, 90);
|
||||
const lookupLimiter = createLimiter(RATE_WINDOW_MS, 60);
|
||||
const ensureLimiter = createLimiter(ENSURE_WINDOW_MS, 3);
|
||||
|
||||
export const isAcceptRateLimited = (ip: string): boolean => acceptLimiter.limited(ip);
|
||||
export const isRelayRateLimited = (peerOrigin: string): boolean => relayLimiter.limited(peerOrigin);
|
||||
export const isLookupRateLimited = (peerOrigin: string): boolean => lookupLimiter.limited(peerOrigin);
|
||||
export const isEnsureRateLimited = (userId: string): boolean => ensureLimiter.limited(userId);
|
||||
|
||||
// Test-only export — used by federation.userLookup.test.ts to reset between cases.
|
||||
export function _resetLookupRateBuckets(): void {
|
||||
lookupLimiter.buckets.clear();
|
||||
}
|
||||
|
||||
// ─── Nonce store for replay protection (per-peer) ────────────────────────────
|
||||
// Maps peerOrigin → (nonce → insertion timestamp). Nonces are evicted after
|
||||
// NONCE_MAX_AGE_MS (15 min) to match the HMAC timestamp window.
|
||||
export const NONCE_MAX_AGE_MS = 15 * 60 * 1000;
|
||||
|
||||
export const nonceStore = new Map<string, Map<string, number>>();
|
||||
|
||||
const NONCE_MAX_AGE_MS = 15 * 60 * 1000;
|
||||
const nonceStore = new Map<string, Map<string, number>>();
|
||||
|
||||
/** Returns true if the nonce is a duplicate (already seen for this peer). */
|
||||
export function isNonceDuplicate(peerOrigin: string, nonce: string): boolean {
|
||||
@@ -146,27 +82,15 @@ export function isNonceDuplicate(peerOrigin: string, nonce: string): boolean {
|
||||
return false;
|
||||
}
|
||||
|
||||
// Periodically clean stale buckets to prevent unbounded memory growth
|
||||
// ─── Periodic cleanup to bound memory ────────────────────────────────────────
|
||||
// Ensure buckets sweep on their own (long) window. Accept + relay buckets and
|
||||
// nonce eviction share the short window. Lookup buckets are pruned per-call only
|
||||
// (never swept here) — preserving the original behavior.
|
||||
setInterval(() => ensureLimiter.sweep(), ENSURE_WINDOW_MS).unref();
|
||||
|
||||
setInterval(() => {
|
||||
const cutoff = Date.now() - ACCEPT_RATE_WINDOW_MS;
|
||||
for (const [ip, timestamps] of acceptRateBuckets) {
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < cutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length === 0) {
|
||||
acceptRateBuckets.delete(ip);
|
||||
}
|
||||
}
|
||||
const relayCutoff = Date.now() - RELAY_RATE_WINDOW_MS;
|
||||
for (const [origin, timestamps] of relayRateBuckets) {
|
||||
while (timestamps.length > 0 && (timestamps[0] ?? Infinity) < relayCutoff) {
|
||||
timestamps.shift();
|
||||
}
|
||||
if (timestamps.length === 0) {
|
||||
relayRateBuckets.delete(origin);
|
||||
}
|
||||
}
|
||||
// Evict expired nonces
|
||||
acceptLimiter.sweep();
|
||||
relayLimiter.sweep();
|
||||
const nonceCutoff = Date.now() - NONCE_MAX_AGE_MS;
|
||||
for (const [origin, nonces] of nonceStore) {
|
||||
for (const [nonce, ts] of nonces) {
|
||||
@@ -174,4 +98,4 @@ setInterval(() => {
|
||||
}
|
||||
if (nonces.size === 0) nonceStore.delete(origin);
|
||||
}
|
||||
}, ACCEPT_RATE_WINDOW_MS).unref();
|
||||
}, RATE_WINDOW_MS).unref();
|
||||
|
||||
Reference in New Issue
Block a user