From 45e1c88bdca7241d02060f4666a8c58afd9a3ecf Mon Sep 17 00:00:00 2001 From: Jannis Braun <151788261+TheZwiss@users.noreply.github.com> Date: Wed, 1 Jul 2026 22:09:43 +0200 Subject: [PATCH] feat(federation): reset detection (markPeerReset) via handshake + probe --- docs/systems/federation.md | 19 ++ docs/systems/websocket.md | 1 + packages/server/src/routes/federation.ts | 25 ++- .../src/utils/federationRecovery.test.ts | 89 +++++++++- .../server/src/utils/federationRecovery.ts | 66 ++++++- .../server/src/utils/federationReset.test.ts | 163 ++++++++++++++++++ packages/server/src/utils/federationReset.ts | 148 ++++++++++++++++ packages/server/src/utils/federationWorker.ts | 14 +- packages/shared/src/types.ts | 1 + 9 files changed, 506 insertions(+), 20 deletions(-) create mode 100644 packages/server/src/utils/federationReset.test.ts create mode 100644 packages/server/src/utils/federationReset.ts diff --git a/docs/systems/federation.md b/docs/systems/federation.md index 655f22f2..d44b5ab9 100644 --- a/docs/systems/federation.md +++ b/docs/systems/federation.md @@ -315,6 +315,24 @@ HMAC-authenticated in **both directions**: the request is signed (only a peer ho **Deterministic epoch-refresh driver (`refreshPeerEpochs()` in `utils/federationEpoch.ts`).** Selects every `active` peer whose `peer_instance_id IS NULL`, calls `fetchPeerEpoch(peer)` once each, and on a non-null result writes the epoch via `UPDATE ... SET peer_instance_id WHERE id = ? AND peer_instance_id IS NULL`. The trailing `IS NULL` guard makes it **populate-if-null only** — it can never overwrite a baseline another path (relay envelope, handshake) already established — and makes it **self-terminating**: once a peer's `peer_instance_id` is set, the `IS NULL` filter excludes it, so it is never fetched again. A `null` from `fetchPeerEpoch` (404 / bad-sig / network) is a benign `continue` with no error log-spam, retried next tick. Wired into the federation worker in two places: once at `startFederationWorkers()` startup and once at the end of `processHealthCheckTick()` (the existing 15-minute health-check tick), both as `refreshPeerEpochs().catch(() => {})`. This guarantees the trusted baseline is populated within one refresh cycle of an upgrade, independent of user/relay activity — the load-bearing populator that relay-only population cannot cover for idle peers. +### Reset Detection (`markPeerReset` — `utils/federationReset.ts`) + +The epoch is a **detection signal only, never an authorization signal.** When a peer behind a known origin advertises an epoch that differs from the trusted baseline (`peer_instance_id`), the instance was wiped and a new incarnation stood up on the same domain. `markPeerReset(peerId, origin, deadEpoch, observedEpoch)` routes the peer for admin attention and snapshots the dead incarnation — but performs **NO rekey, NO tombstone, NO handle change, NO content deletion.** The actual data heal fires only later, from `onPeerActivated` after an admin-authenticated re-peer (design §6). Because detection grants no capability and destroys nothing, it is safe to fire on an unauthenticated signal: the worst a spoofed detection can do is flag a peer for admin review (admin-reversible nuisance). + +In a single transaction, `markPeerReset`: +1. Sets the peer `status='needs_attention'`, `needs_attention_reason='peer_reset_detected'`, and `observed_peer_instance_id=observedEpoch`. **`peer_instance_id` (the trusted baseline) and `hmac_secret` are left untouched** — an unauthenticated observation never rekeys trust; the observed-but-untrusted epoch lives only in `observed_peer_instance_id`. +2. **Snapshots the dead incarnation:** sets `users.federation_heal_pending = 1` for every non-deleted user whose `home_instance` matches the origin. The match keys on `extractDomain(origin)` (bare domain, the canonical `home_instance` form) and defensively also matches the `https://`/`http://`-prefixed forms so any legacy full-URL straggler is caught (`homeInstanceMatch()`). Any stub created *after* detection (e.g. a friend-add reaching the new incarnation directly) is un-flagged and survives the heal. +3. **Journals the dead incarnation durably** by upserting a `federation_reset_events` row keyed by origin: `{ dead_epoch=deadEpoch, new_epoch=NULL, detected_at, resolved_at=NULL, stub_count, orphaned_account_count }`. `stub_count` counts flagged pure S2S stubs (`password_hash = '!federation-replicated'`); `orphaned_account_count` counts flagged real accounts. This row survives the peer-row deletion that Re-peer performs, preserving `dead_epoch` for the false-positive guard (design §6.1) and the admin surface. +4. After the transaction, broadcasts `federation_peers_changed` and `federation_peer_reset_detected {origin}` to admins. + +**Idempotent / double-reset:** if an *unresolved* `federation_reset_events` row already exists for the origin (the peer reset again before an admin resolved the first), the original `dead_epoch` and `detected_at` are **preserved** (that is the incarnation whose users are already snapshotted) — only the summary counts are refreshed. `dead_epoch` is never overwritten on an unresolved row. A prior *resolved* reset starts a fresh journal entry. + +**Detection sources (both wired in this feature):** +- **Inbound handshake** — `/peer/accept` landing on an `active`/`needs_attention` row (`routes/federation.ts`): before the idempotent-200 return, `if (reqInstanceId && existing.peerInstanceId && reqInstanceId !== existing.peerInstanceId) markPeerReset(...)`. The guard still returns 200 and **still does not rekey** — the anti-hijack property is preserved verbatim; detection is layered on top. +- **Reachability probe** — `probePeerReachable()` (`utils/federationRecovery.ts`) now parses `instanceId` from the `/api/instance/info` response (returning `{ reachable, instanceId }`; a missing/unparseable epoch is `null`, never an error). The shared decision helper `recoverOrDetectReset(peer, result)` — used by both the background recovery tick (`processRecoveryTick`) and the manual recheck endpoint — routes to `markPeerReset` (returning `'reset_detected'`) when the peer has a non-null `peer_instance_id` and the probed epoch differs, and **does NOT call `markPeerRecovered`**. Rationale: a genuinely reset peer's HMAC secret is desynced, so flipping it back to `active` via a reachability probe would resume relay against a dead secret. Only when the probed epoch matches the baseline (or the baseline is null / epoch unknown) does the normal recovery-to-active path run. The manual recheck endpoint returns `{ recovered: false, status: 'needs_attention' }` on `'reset_detected'`. + +Legacy peers advertise no epoch (`instanceId` null), so detection requires a non-null observed epoch differing from a non-null stored baseline — legacy peers never trigger it, and the existing `auth_failures → needs_attention → manual Reset` path continues unchanged for them. + ### S2S Identity Deletion (`DELETE /api/federation/identity`) Allows a home instance to remove a user's replicated identity from a remote instance. @@ -357,6 +375,7 @@ These S→C events are pushed to the acting user's connected clients by the fede |-------|-------------|---------| | `federation_peer_rejected` | Outbox worker receives `403 PEERING_REQUIRES_APPROVAL` from a remote instance during auto-peering | `{ peerId: string, origin: string }` | | `federation_peer_active` | A previously `rejected` peer transitions to `active` (e.g., via manual `peer/initiate` or incoming `peer/accept`) | `{ peerId: string, origin: string }` | +| `federation_peer_reset_detected` | A peer's advertised instance epoch differs from the trusted baseline (wipe-and-reinstall on the same domain) — emitted by `markPeerReset` after routing the peer to `needs_attention` | `{ origin: string }` (admin-only, via `sendToAdmins`) | --- diff --git a/docs/systems/websocket.md b/docs/systems/websocket.md index bd0ae788..911be66c 100644 --- a/docs/systems/websocket.md +++ b/docs/systems/websocket.md @@ -194,6 +194,7 @@ reason: `'displaced'` (new tab) | `'session_closed'` |------|--------|-------| | `federation_file_rejected` | messageId, dmChannelId, attachmentId, affectedUsers[] | DM members | | `federation_approval_request_received` | — (refetch trigger; payload: `{ type }`) | admins. Fires for **both** inbound peering requests (remote → us) AND outbound queue creation when the [Outbound Peering Gate](federation.md#outbound-peering-gate) creates a `peer_approval_requests` row in response to a user_action. Payload shape unchanged from the inbound-only behavior; only the firing surface widened. | +| `federation_peer_reset_detected` | `{ origin: string }` | admins. Fires from `markPeerReset` when a peer's advertised instance epoch differs from the trusted baseline (a wipe-and-reinstall on the same domain — see [Reset Detection](federation.md#reset-detection-markpeerreset--utilsfederationresetts)). Detection-only: the peer was routed to `needs_attention` (reason `peer_reset_detected`) with no rekey/tombstone. Paired with a `federation_peers_changed` broadcast; client surfaces the reset for one-click Re-peer. | | `peering_subscription_changed` | — (refetch trigger; payload: `{ type }`) | the subscribing user (all of their connected sessions). Fires when a `peer_approval_subscribers` row belonging to the user is created, modified, or deleted (gate fan-in, user cancel, parent cascade). Client refetches `GET /api/federation/peering-subscriptions`. | | `peering_notification_received` | `{ type, kind: 'approved' \| 'denied' \| 'expired' }` | the user the notification belongs to. Fires when a `peer_approval_notifications` row is created (`onPeerActivated` outbound fanout, outbound `/deny` fanout, janitor outbound expiry). Client refetches `GET /api/federation/peering-notifications` and may surface a transient toast for online users. | diff --git a/packages/server/src/routes/federation.ts b/packages/server/src/routes/federation.ts index 40fc2152..de102c51 100644 --- a/packages/server/src/routes/federation.ts +++ b/packages/server/src/routes/federation.ts @@ -20,7 +20,8 @@ import { tombstoneUser, collectDeletionBroadcastTargets, collectProfileBroadcast import { computeFederatedId, getDmParticipants, sendCallRelay } from '../utils/federationOutbox.js'; import { onPeerActivated, onPeerDeactivated } from '../utils/federationPeerActivation.js'; import { getInstanceId } from '../utils/federationEpoch.js'; -import { probePeerReachable, markPeerRecovered } from '../utils/federationRecovery.js'; +import { probePeerReachable, recoverOrDetectReset } from '../utils/federationRecovery.js'; +import { markPeerReset } from '../utils/federationReset.js'; import { getDmMessageWithUser } from './dm.js'; import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent, FederationRelayAttachment, FederationSyncRequest, FederationSyncResponse, DmMessageWithUser, DmChannel, FederationRelayProfileSnapshot, FederationIdentityDeleteS2SRequest, FederationProfileUpdatePayload, ServerEvent, ApprovalRequestSubscriberSummary, PeeringTriggerReason } from '@backspace/shared'; import { GROUP_DM_NAME_MIN_LENGTH, GROUP_DM_NAME_MAX_LENGTH } from '@backspace/shared/src/constants.js'; @@ -1120,6 +1121,16 @@ export async function federationRoutes(app: FastifyInstance): Promise { // Legitimate recovery path: local admin clicks "Reset peering" → // row is deleted → remote's /peer/accept then lands on a // non-existent row and the normal handshake path runs. + // + // Detection-only: if the inbound epoch differs from our trusted + // baseline, the peer is a NEW incarnation on the same domain (a + // wipe-and-reinstall). Route it to needs_attention + snapshot + + // journal — but STILL return 200 and STILL do not rekey. The + // anti-hijack guard above is preserved verbatim; detection never + // grants capability. + if (reqInstanceId && existing.peerInstanceId && reqInstanceId !== existing.peerInstanceId) { + markPeerReset(existing.id, sourceOrigin, existing.peerInstanceId, reqInstanceId); + } return reply.code(200).send({ accepted: true, instanceName: ourInstanceName, instanceId: ourInstanceId }); } if (existing.status === 'revoked') { @@ -1619,10 +1630,16 @@ export async function federationRoutes(app: FastifyInstance): Promise { }); } - const reachable = await probePeerReachable(peer.origin); + const probe = await probePeerReachable(peer.origin); - if (reachable) { - await markPeerRecovered(peer.id); + if (probe.reachable) { + const outcome = await recoverOrDetectReset(peer, probe); + if (outcome === 'reset_detected') { + // The peer is a new incarnation on the same domain. It was routed to + // needs_attention (detection-only, no rekey) and must NOT be recovered + // to active until an admin re-peers through the authenticated path. + return reply.code(200).send({ recovered: false, status: 'needs_attention' }); + } return reply.code(200).send({ recovered: true, status: 'active' }); } diff --git a/packages/server/src/utils/federationRecovery.test.ts b/packages/server/src/utils/federationRecovery.test.ts index fea143cd..db91d1e1 100644 --- a/packages/server/src/utils/federationRecovery.test.ts +++ b/packages/server/src/utils/federationRecovery.test.ts @@ -17,6 +17,11 @@ vi.mock('../db/index.js', () => ({ getDb: () => testDb, schema })); const onPeerActivated = vi.fn(); vi.mock('./federationPeerActivation.js', () => ({ onPeerActivated })); +const sendToAdmins = vi.fn(); +vi.mock('../ws/handler.js', () => ({ + connectionManager: { sendToAdmins, getAllOnlineUserIds: () => [], sendToUser: vi.fn() }, +})); + function applyMigrations(db: Database.Database): void { const dir = path.resolve(__dirname, '../../drizzle'); for (const f of fs.readdirSync(dir).filter(x => x.endsWith('.sql')).sort()) { @@ -45,23 +50,35 @@ describe('federationRecovery primitives', () => { vi.clearAllMocks(); }); - it('probePeerReachable returns true on a 200 from /api/instance/info', async () => { - const spy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{}', { status: 200 })); + it('probePeerReachable returns reachable + parsed instanceId on a 200 from /api/instance/info', async () => { + const spy = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{"instanceId":"epoch-x"}', { status: 200 })); const { probePeerReachable } = await import('./federationRecovery.js'); - await expect(probePeerReachable('https://peer.example')).resolves.toBe(true); + await expect(probePeerReachable('https://peer.example')).resolves.toEqual({ reachable: true, instanceId: 'epoch-x' }); expect(spy).toHaveBeenCalledWith('https://peer.example/api/instance/info', expect.anything()); }); - it('probePeerReachable returns false on a non-ok response', async () => { - vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 502 })); + it('probePeerReachable reports null instanceId when the body omits it (legacy peer)', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('{}', { status: 200 })); const { probePeerReachable } = await import('./federationRecovery.js'); - await expect(probePeerReachable('https://peer.example')).resolves.toBe(false); + await expect(probePeerReachable('https://peer.example')).resolves.toEqual({ reachable: true, instanceId: null }); }); - it('probePeerReachable returns false on network error', async () => { + it('probePeerReachable reports null instanceId when the body is unparseable', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('not-json', { status: 200 })); + const { probePeerReachable } = await import('./federationRecovery.js'); + await expect(probePeerReachable('https://peer.example')).resolves.toEqual({ reachable: true, instanceId: null }); + }); + + it('probePeerReachable returns not-reachable on a non-ok response', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('', { status: 502 })); + const { probePeerReachable } = await import('./federationRecovery.js'); + await expect(probePeerReachable('https://peer.example')).resolves.toEqual({ reachable: false, instanceId: null }); + }); + + it('probePeerReachable returns not-reachable on network error', async () => { vi.spyOn(globalThis, 'fetch').mockRejectedValue(new Error('ENOTFOUND')); const { probePeerReachable } = await import('./federationRecovery.js'); - await expect(probePeerReachable('https://peer.example')).resolves.toBe(false); + await expect(probePeerReachable('https://peer.example')).resolves.toEqual({ reachable: false, instanceId: null }); }); it('markPeerRecovered flips status to active, resets pacing + counters, calls onPeerActivated', async () => { @@ -77,4 +94,60 @@ describe('federationRecovery primitives', () => { expect(row.lastSeenAt).toBeGreaterThan(0); expect(onPeerActivated).toHaveBeenCalledWith('peer-rec', 'health_check_recovery'); }); + + it('recoverOrDetectReset recovers when the probed epoch matches the trusted baseline', async () => { + testDb.insert(schema.federationPeers).values({ + id: 'peer-match', origin: 'https://peer.example', hmacSecret: 'secret', + status: 'unreachable', consecutiveFailures: 5, probeAttempts: 2, lastProbeAt: Date.now(), + peerInstanceId: 'E0', lastSyncedAt: Date.now(), createdAt: Date.now(), + }).run(); + const { recoverOrDetectReset } = await import('./federationRecovery.js'); + const outcome = await recoverOrDetectReset( + { id: 'peer-match', origin: 'https://peer.example', peerInstanceId: 'E0' }, + { reachable: true, instanceId: 'E0' }, + ); + expect(outcome).toBe('recovered'); + const row = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-match')).get()!; + expect(row.status).toBe('active'); + expect(onPeerActivated).toHaveBeenCalledWith('peer-match', 'health_check_recovery'); + }); + + it('recoverOrDetectReset recovers when the baseline is null (never-tracked / legacy)', async () => { + testDb.insert(schema.federationPeers).values({ + id: 'peer-null', origin: 'https://peer.example', hmacSecret: 'secret', + status: 'unreachable', consecutiveFailures: 5, probeAttempts: 2, lastProbeAt: Date.now(), + peerInstanceId: null, lastSyncedAt: Date.now(), createdAt: Date.now(), + }).run(); + const { recoverOrDetectReset } = await import('./federationRecovery.js'); + const outcome = await recoverOrDetectReset( + { id: 'peer-null', origin: 'https://peer.example', peerInstanceId: null }, + { reachable: true, instanceId: 'E9' }, + ); + expect(outcome).toBe('recovered'); + expect(testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-null')).get()!.status).toBe('active'); + }); + + it('recoverOrDetectReset routes to needs_attention (does NOT recover) when the probed epoch differs', async () => { + testDb.insert(schema.federationPeers).values({ + id: 'peer-reset', origin: 'https://peer.example', hmacSecret: 'secret', + status: 'unreachable', consecutiveFailures: 5, probeAttempts: 2, lastProbeAt: Date.now(), + peerInstanceId: 'E0', lastSyncedAt: Date.now(), createdAt: Date.now(), + }).run(); + const { recoverOrDetectReset } = await import('./federationRecovery.js'); + const outcome = await recoverOrDetectReset( + { id: 'peer-reset', origin: 'https://peer.example', peerInstanceId: 'E0' }, + { reachable: true, instanceId: 'E1' }, + ); + expect(outcome).toBe('reset_detected'); + const row = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-reset')).get()!; + expect(row.status).toBe('needs_attention'); + expect(row.needsAttentionReason).toBe('peer_reset_detected'); + expect(row.peerInstanceId).toBe('E0'); // trusted baseline untouched + expect(row.hmacSecret).toBe('secret'); // never rekeyed + // A reset peer must NOT be recovered to active. + expect(onPeerActivated).not.toHaveBeenCalled(); + }); }); diff --git a/packages/server/src/utils/federationRecovery.ts b/packages/server/src/utils/federationRecovery.ts index 1bfffef7..c9734173 100644 --- a/packages/server/src/utils/federationRecovery.ts +++ b/packages/server/src/utils/federationRecovery.ts @@ -2,25 +2,55 @@ import { getDb } from '../db/index.js'; import * as schema from '../db/schema.js'; import { eq } from 'drizzle-orm'; import { onPeerActivated } from './federationPeerActivation.js'; +import { markPeerReset } from './federationReset.js'; /** Reachability-probe timeout (ms). */ export const RECOVERY_PROBE_TIMEOUT_MS = 10_000; +/** + * Result of a reachability probe. `instanceId` is the peer's advertised instance + * epoch (from `/api/instance/info`), used for reset detection. It is `null` when + * the peer is unreachable, when it is too old to advertise an epoch, or when the + * body is unparseable — all of which degrade to "no reset observed." + */ +export interface ProbeResult { + reachable: boolean; + instanceId: string | null; +} + /** * Liveness probe shared by the recovery tick and the manual recheck endpoint. * GET {origin}/api/instance/info with a 10s timeout. No HMAC — reachability is * not trust; a recovered-but-HMAC-broken peer still transitions to * needs_attention via the auth-failure path on the next real delivery. + * + * Also parses the peer's advertised `instanceId` (instance epoch) from the + * response so callers can detect a wipe-and-reinstall (a NEW incarnation on the + * same domain). A missing/unparseable epoch is reported as `null` — never an + * error — so a legacy peer that omits it simply recovers normally. */ -export async function probePeerReachable(origin: string, signal?: AbortSignal): Promise { +export async function probePeerReachable(origin: string, signal?: AbortSignal): Promise { try { const timeout = AbortSignal.timeout(RECOVERY_PROBE_TIMEOUT_MS); const response = await fetch(`${origin}/api/instance/info`, { signal: signal ? AbortSignal.any([signal, timeout]) : timeout, }); - return response.ok; + if (!response.ok) { + return { reachable: false, instanceId: null }; + } + let instanceId: string | null = null; + try { + const body = (await response.json()) as { instanceId?: unknown }; + if (typeof body?.instanceId === 'string' && body.instanceId.length > 0) { + instanceId = body.instanceId; + } + } catch { + // Reachable but body unparseable — treat epoch as unknown, not a failure. + instanceId = null; + } + return { reachable: true, instanceId }; } catch { - return false; + return { reachable: false, instanceId: null }; } } @@ -43,3 +73,33 @@ export async function markPeerRecovered(peerId: string): Promise { .run(); await onPeerActivated(peerId, 'health_check_recovery'); } + +/** + * Decide the outcome of a successful reachability probe for a peer that is + * eligible to recover. This is the single recovery-decision point shared by the + * background recovery worker and the manual recheck endpoint. + * + * Detection-only reset gate: if the peer has a trusted baseline epoch + * (`peer_instance_id`) AND the probe observed a DIFFERENT epoch, the peer is a + * new incarnation on the same domain. A genuinely reset peer's HMAC secret is + * desynced, so flipping it back to `active` via a reachability probe would + * resume relay against a dead secret. We therefore route it to + * `needs_attention` via `markPeerReset` and DO NOT recover it — it must wait for + * an admin-authenticated re-handshake. Only when the epoch matches the baseline + * (or the baseline is null / the epoch is unknown) does the normal recovery path + * run. + * + * @returns `'reset_detected'` if the peer was routed to needs_attention; + * `'recovered'` if it was flipped back to active. + */ +export async function recoverOrDetectReset( + peer: { id: string; origin: string; peerInstanceId: string | null }, + result: ProbeResult, +): Promise<'recovered' | 'reset_detected'> { + if (peer.peerInstanceId && result.instanceId && result.instanceId !== peer.peerInstanceId) { + markPeerReset(peer.id, peer.origin, peer.peerInstanceId, result.instanceId); + return 'reset_detected'; + } + await markPeerRecovered(peer.id); + return 'recovered'; +} diff --git a/packages/server/src/utils/federationReset.test.ts b/packages/server/src/utils/federationReset.test.ts new file mode 100644 index 00000000..b0396668 --- /dev/null +++ b/packages/server/src/utils/federationReset.test.ts @@ -0,0 +1,163 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'; +import Database from 'better-sqlite3'; +import { drizzle } from 'drizzle-orm/better-sqlite3'; +import fs from 'node:fs'; +import path from 'node:path'; +import { fileURLToPath } from 'node:url'; +import * as schema from '../db/schema.js'; +import { eq } from 'drizzle-orm'; + +const __dirname = path.dirname(fileURLToPath(import.meta.url)); +type TestDb = ReturnType>; +let sqlite: Database.Database; +let testDb: TestDb; + +vi.mock('../db/index.js', () => ({ getDb: () => testDb, getRawDb: () => sqlite, schema })); + +const sendToAdmins = vi.fn(); +vi.mock('../ws/handler.js', () => ({ + connectionManager: { sendToAdmins, getAllOnlineUserIds: () => [], sendToUser: vi.fn() }, +})); + +const STUB = '!federation-replicated'; +const ORIGIN = 'https://peer.example'; +const DOMAIN = 'peer.example'; + +function applyMigrations(db: Database.Database): void { + const dir = path.resolve(__dirname, '../../drizzle'); + for (const f of fs.readdirSync(dir).filter(x => x.endsWith('.sql')).sort()) { + const sqlText = fs.readFileSync(path.join(dir, f), 'utf8'); + for (const stmt of sqlText.split(/-->\s*statement-breakpoint/)) { + const clean = stmt.trim(); + if (clean) db.exec(clean); + } + } +} + +function seedPeer(): void { + testDb.insert(schema.federationPeers).values({ + id: 'peer-1', origin: ORIGIN, hmacSecret: 'trusted-secret', + status: 'active', peerInstanceId: 'E0', lastSeenAt: Date.now(), + lastSyncedAt: Date.now(), createdAt: Date.now(), + }).run(); +} + +function seedUser(id: string, opts: { passwordHash: string; isDeleted?: number; homeInstance?: string }): void { + testDb.insert(schema.users).values({ + id, username: `${id}@${DOMAIN}`, passwordHash: opts.passwordHash, + homeInstance: opts.homeInstance ?? DOMAIN, homeUserId: id, + isDeleted: opts.isDeleted ?? 0, createdAt: Date.now(), + }).run(); +} + +describe('markPeerReset — detection-only reset routing', () => { + beforeEach(() => { + sqlite = new Database(':memory:'); + testDb = drizzle(sqlite, { schema }); + applyMigrations(sqlite); + vi.clearAllMocks(); + }); + + afterEach(() => { + sqlite.close(); + }); + + it('routes peer to needs_attention, snapshots the dead incarnation, and journals the dead epoch', async () => { + seedPeer(); + // Two non-deleted users for the reset origin: one pure S2S stub, one real account. + seedUser('stub-1', { passwordHash: STUB }); + seedUser('real-1', { passwordHash: '$2b$10$realbcrypthash' }); + // A deleted stub for the same origin — must NOT be flagged. + seedUser('stub-deleted', { passwordHash: STUB, isDeleted: 1 }); + // An unrelated user on a different origin — must NOT be flagged. + seedUser('other-1', { passwordHash: STUB, homeInstance: 'elsewhere.example' }); + + const { markPeerReset } = await import('./federationReset.js'); + markPeerReset('peer-1', ORIGIN, 'E0', 'E1'); + + const peer = testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-1')).get()!; + expect(peer.status).toBe('needs_attention'); + expect(peer.needsAttentionReason).toBe('peer_reset_detected'); + expect(peer.observedPeerInstanceId).toBe('E1'); + // Trusted baseline + secret are NEVER touched by detection. + expect(peer.peerInstanceId).toBe('E0'); + expect(peer.hmacSecret).toBe('trusted-secret'); + + const flag = (id: string) => testDb.select().from(schema.users) + .where(eq(schema.users.id, id)).get()!.federationHealPending; + expect(flag('stub-1')).toBe(1); + expect(flag('real-1')).toBe(1); + expect(flag('stub-deleted')).toBe(0); // deleted → excluded from snapshot + expect(flag('other-1')).toBe(0); // different origin → excluded + + const journal = testDb.select().from(schema.federationResetEvents) + .where(eq(schema.federationResetEvents.origin, ORIGIN)).get()!; + expect(journal.deadEpoch).toBe('E0'); + expect(journal.newEpoch).toBeNull(); + expect(journal.resolvedAt).toBeNull(); + expect(journal.stubCount).toBe(1); // stub-1 only (deleted stub excluded) + expect(journal.orphanedAccountCount).toBe(1); // real-1 + + // Admin broadcast fired. + expect(sendToAdmins).toHaveBeenCalledWith({ type: 'federation_peers_changed' }); + expect(sendToAdmins).toHaveBeenCalledWith({ type: 'federation_peer_reset_detected', origin: ORIGIN }); + }); + + it('double-reset keeps the original dead_epoch and detected_at (only counts refresh)', async () => { + seedPeer(); + seedUser('stub-1', { passwordHash: STUB }); + seedUser('real-1', { passwordHash: '$2b$10$realbcrypthash' }); + + const { markPeerReset } = await import('./federationReset.js'); + markPeerReset('peer-1', ORIGIN, 'E0', 'E1'); + const first = testDb.select().from(schema.federationResetEvents) + .where(eq(schema.federationResetEvents.origin, ORIGIN)).get()!; + const originalDetectedAt = first.detectedAt; + + // Peer resets AGAIN before an admin resolved the first reset. + markPeerReset('peer-1', ORIGIN, 'E0', 'E2'); + const second = testDb.select().from(schema.federationResetEvents) + .where(eq(schema.federationResetEvents.origin, ORIGIN)).get()!; + + // The dead epoch is the ALREADY-snapshotted incarnation — never overwritten. + expect(second.deadEpoch).toBe('E0'); + expect(second.detectedAt).toBe(originalDetectedAt); + expect(second.resolvedAt).toBeNull(); + // The observed epoch on the peer row does advance to the newest observation. + expect(testDb.select().from(schema.federationPeers) + .where(eq(schema.federationPeers.id, 'peer-1')).get()!.observedPeerInstanceId).toBe('E2'); + }); + + it('a resolved prior reset starts a fresh journal entry on a new reset', async () => { + seedPeer(); + seedUser('stub-1', { passwordHash: STUB }); + + const { markPeerReset } = await import('./federationReset.js'); + markPeerReset('peer-1', ORIGIN, 'E0', 'E1'); + // Simulate the heal having resolved the first reset. + testDb.update(schema.federationResetEvents) + .set({ resolvedAt: Date.now(), newEpoch: 'E1' }) + .where(eq(schema.federationResetEvents.origin, ORIGIN)).run(); + + // A brand-new reset lands: dead_epoch should update to the new baseline. + markPeerReset('peer-1', ORIGIN, 'E1', 'E2'); + const row = testDb.select().from(schema.federationResetEvents) + .where(eq(schema.federationResetEvents.origin, ORIGIN)).get()!; + expect(row.deadEpoch).toBe('E1'); + expect(row.newEpoch).toBeNull(); + expect(row.resolvedAt).toBeNull(); + }); + + it('matches home_instance stored as a full URL (defensive format match)', async () => { + seedPeer(); + // Legacy straggler stored with the https:// prefix rather than bare domain. + seedUser('stub-url', { passwordHash: STUB, homeInstance: ORIGIN }); + + const { markPeerReset } = await import('./federationReset.js'); + markPeerReset('peer-1', ORIGIN, 'E0', 'E1'); + + expect(testDb.select().from(schema.users) + .where(eq(schema.users.id, 'stub-url')).get()!.federationHealPending).toBe(1); + }); +}); diff --git a/packages/server/src/utils/federationReset.ts b/packages/server/src/utils/federationReset.ts new file mode 100644 index 00000000..f05fa706 --- /dev/null +++ b/packages/server/src/utils/federationReset.ts @@ -0,0 +1,148 @@ +import { and, eq, sql } from 'drizzle-orm'; +import { getDb, schema } from '../db/index.js'; +import { extractDomain } from '../routes/federation.js'; +import { connectionManager } from '../ws/handler.js'; + +/** Pure-stub sentinel: a user replicated purely over S2S (no local credentials). */ +const REPLICATED_STUB_SENTINEL = '!federation-replicated'; + +/** + * SQL predicate matching every local user whose home instance is `origin`. + * + * `users.home_instance` is stored canonically as a bare domain + * (`resolveOrCreateReplicatedUser` writes `extractDomain(...)`), so we key on + * the bare domain. We additionally match the `https://`/`http://`-prefixed + * forms so any legacy full-URL straggler is still caught — mirroring the + * defensive normalization used across the outbox/worker paths. A silent + * zero-match here would no-op the entire heal, so the match is deliberately + * permissive on format while exact on domain. + */ +export function homeInstanceMatch(origin: string) { + const domain = extractDomain(origin); + return sql`(${schema.users.homeInstance} = ${domain} OR ${schema.users.homeInstance} = ${'https://' + domain} OR ${schema.users.homeInstance} = ${'http://' + domain})`; +} + +/** + * Detection-only reset routing. Invoked when a peer behind a known origin is + * observed to carry a DIFFERENT instance epoch than the trusted baseline + * (`federation_peers.peer_instance_id`) — i.e. the instance was wiped and a new + * incarnation stood up on the same domain. + * + * This routes the peer to `needs_attention` (reason `peer_reset_detected`), + * snapshots the dead incarnation's users (`federation_heal_pending = 1`), and + * journals the dead epoch durably in `federation_reset_events`. It then notifies + * admins. + * + * It performs **NO rekey, NO tombstone, NO handle change, NO content deletion**. + * The trusted baseline (`peer_instance_id`) and the `hmac_secret` are left + * untouched — the observed (but not yet trusted) epoch is recorded separately in + * `observed_peer_instance_id`. Trust re-establishment is admin-gated (§5) and + * the actual data heal fires only after an authenticated re-peer (§6). Because + * none of this grants capability or destroys content, it is safe to fire on an + * unauthenticated detection signal: the worst a spoofed detection can do is flag + * a peer for admin review. + * + * Idempotent: if an UNRESOLVED reset row already exists for the origin (the peer + * reset again before an admin resolved the first), the original `dead_epoch` and + * `detected_at` are preserved — that is the incarnation whose users are already + * snapshotted — and only the summary counts are refreshed. + * + * @param peerId `federation_peers.id` of the reset peer. + * @param origin The peer origin (bare domain or full URL). + * @param deadEpoch The peer's trusted baseline epoch at detection time. + * @param observedEpoch The new epoch observed on the peer. + */ +export function markPeerReset(peerId: string, origin: string, deadEpoch: string, observedEpoch: string): void { + const db = getDb(); + + db.transaction((tx) => { + // 1. Route the peer to needs_attention and record the observed (untrusted) + // epoch. peer_instance_id (trusted baseline) and hmac_secret are NOT + // touched — an unauthenticated observation never rekeys trust. + tx.update(schema.federationPeers) + .set({ + status: 'needs_attention', + needsAttentionReason: 'peer_reset_detected', + observedPeerInstanceId: observedEpoch, + }) + .where(eq(schema.federationPeers.id, peerId)) + .run(); + + // 2. Snapshot exactly the current (dead-incarnation) users for this origin. + // Any stub created AFTER this point (e.g. a friend-add reaching the new + // incarnation directly) is un-flagged and survives the heal. + tx.update(schema.users) + .set({ federationHealPending: 1 }) + .where(and(eq(schema.users.isDeleted, 0), homeInstanceMatch(origin))) + .run(); + + // 3. Compute summary counts for the admin surface, over the freshly-flagged + // set: pure replicated stubs vs. real federated accounts (local content). + const stubCount = tx + .select({ n: sql`count(*)` }) + .from(schema.users) + .where(and( + eq(schema.users.federationHealPending, 1), + eq(schema.users.passwordHash, REPLICATED_STUB_SENTINEL), + homeInstanceMatch(origin), + )) + .get()?.n ?? 0; + + const orphanedAccountCount = tx + .select({ n: sql`count(*)` }) + .from(schema.users) + .where(and( + eq(schema.users.federationHealPending, 1), + sql`${schema.users.passwordHash} != ${REPLICATED_STUB_SENTINEL}`, + homeInstanceMatch(origin), + )) + .get()?.n ?? 0; + + // 4. Journal the dead incarnation durably. This row survives the peer-row + // deletion that Re-peer performs, preserving dead_epoch for the + // false-positive guard and the admin surface. + const existing = tx + .select() + .from(schema.federationResetEvents) + .where(eq(schema.federationResetEvents.origin, origin)) + .get(); + + if (existing && existing.resolvedAt === null) { + // Double-reset: keep the ORIGINAL dead_epoch + detected_at (the + // incarnation already snapshotted), refresh counts only. Never overwrite + // dead_epoch on an unresolved row. + tx.update(schema.federationResetEvents) + .set({ stubCount, orphanedAccountCount }) + .where(eq(schema.federationResetEvents.origin, origin)) + .run(); + } else { + // First detection for this origin, or a prior reset that was already + // resolved — start a fresh journal entry. + tx.insert(schema.federationResetEvents) + .values({ + origin, + deadEpoch, + newEpoch: null, + detectedAt: Date.now(), + resolvedAt: null, + stubCount, + orphanedAccountCount, + }) + .onConflictDoUpdate({ + target: schema.federationResetEvents.origin, + set: { + deadEpoch, + newEpoch: null, + detectedAt: Date.now(), + resolvedAt: null, + stubCount, + orphanedAccountCount, + }, + }) + .run(); + } + }); + + connectionManager.sendToAdmins({ type: 'federation_peers_changed' as const }); + connectionManager.sendToAdmins({ type: 'federation_peer_reset_detected' as const, origin }); +} diff --git a/packages/server/src/utils/federationWorker.ts b/packages/server/src/utils/federationWorker.ts index 7d6ffc8d..ecfbd02b 100644 --- a/packages/server/src/utils/federationWorker.ts +++ b/packages/server/src/utils/federationWorker.ts @@ -12,7 +12,7 @@ import { connectionManager } from '../ws/handler.js'; import { generateThumbnail } from './thumbnail.js'; import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared'; import { startupBootstrapSync, onPeerDeactivated } from './federationPeerActivation.js'; -import { probePeerReachable, markPeerRecovered } from './federationRecovery.js'; +import { probePeerReachable, recoverOrDetectReset } from './federationRecovery.js'; import { backfillReplicatedProfileAssets } from '../routes/federation.js'; import { invokePermanentFailureCallback } from './federationRollback.js'; import { refreshPeerEpochs, getInstanceId } from './federationEpoch.js'; @@ -1057,11 +1057,15 @@ export async function processRecoveryTick(): Promise { if (!due) continue; recoveryAbortController = new AbortController(); - const reachable = await probePeerReachable(peer.origin, recoveryAbortController.signal); + const probe = await probePeerReachable(peer.origin, recoveryAbortController.signal); - if (reachable) { - await markPeerRecovered(peer.id); - console.log(`[federation-worker] Peer ${peer.origin} recovered — marked active`); + if (probe.reachable) { + const outcome = await recoverOrDetectReset(peer, probe); + if (outcome === 'reset_detected') { + console.warn(`[federation-worker] Peer ${peer.origin} reset detected (new instance epoch) — routed to needs_attention`); + } else { + console.log(`[federation-worker] Peer ${peer.origin} recovered — marked active`); + } } else { db.update(schema.federationPeers) .set({ probeAttempts: peer.probeAttempts + 1, lastProbeAt: now }) diff --git a/packages/shared/src/types.ts b/packages/shared/src/types.ts index aa2796a7..066157ea 100644 --- a/packages/shared/src/types.ts +++ b/packages/shared/src/types.ts @@ -486,6 +486,7 @@ export type ServerEvent = | { type: 'federation_peer_rejected'; peerOrigin: string; peerLabel?: string; reason: string; affectedContexts: Array<{ contextType: 'dm' | 'friend'; contextId: string; contextLabel: string }> } | { type: 'federation_peer_active'; peerOrigin: string } | { type: 'federation_peers_changed' } + | { type: 'federation_peer_reset_detected'; origin: string } | { type: 'federation_approval_request_received'; origin: string; instanceName?: string } | { type: 'peering_subscription_changed' } | { type: 'peering_notification_received'; kind: PeeringNotificationKind }