feat(federation): reset detection (markPeerReset) via handshake + probe
This commit is contained in:
Jannis Braun
@@ -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`) |
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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. |
|
||||
|
||||
|
||||
@@ -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<void> {
|
||||
// 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<void> {
|
||||
});
|
||||
}
|
||||
|
||||
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' });
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<boolean> {
|
||||
export async function probePeerReachable(origin: string, signal?: AbortSignal): Promise<ProbeResult> {
|
||||
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<void> {
|
||||
.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';
|
||||
}
|
||||
|
||||
@@ -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<typeof drizzle<typeof schema>>;
|
||||
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);
|
||||
});
|
||||
});
|
||||
@@ -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<number>`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<number>`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 });
|
||||
}
|
||||
@@ -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<void> {
|
||||
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 })
|
||||
|
||||
@@ -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 }
|
||||
|
||||
Reference in New Issue
Block a user