diff --git a/app/api/plugin-proxy/route.ts b/app/api/plugin-proxy/route.ts index 2715160e7..979eb4c26 100644 --- a/app/api/plugin-proxy/route.ts +++ b/app/api/plugin-proxy/route.ts @@ -5,6 +5,7 @@ import { rejectCrossOriginRequest } from '@/lib/security/same-origin'; import { getStalwartCredentials } from '@/lib/stalwart/credentials'; import { CREDENTIAL_ID_RE } from '@/lib/plugin-secrets/credentials'; import { ProxyError, proxyRequest, type CaptureSpec } from '@/lib/plugin-secrets/proxy'; +import { bootGateResponse } from '@/lib/shared-store/boot-gate'; const PLUGIN_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/; const FIELD_RE = /^[A-Za-z_][A-Za-z0-9_]{0,63}$/; @@ -35,6 +36,9 @@ function parseCapture(raw: string | null): CaptureSpec | null | undefined { * plugin itself (its iframe has an opaque origin and is refused here). */ export async function POST(request: NextRequest) { + // The proxy doesn't run for this route, so it applies the boot gate itself. + const notReady = bootGateResponse(); + if (notReady) return notReady; const crossOrigin = rejectCrossOriginRequest(request); if (crossOrigin) return crossOrigin; diff --git a/app/api/ready/route.ts b/app/api/ready/route.ts new file mode 100644 index 000000000..9bb71294a --- /dev/null +++ b/app/api/ready/route.ts @@ -0,0 +1,55 @@ +import { NextResponse } from 'next/server'; +import { bootState, getStore, replicaId } from '@/lib/shared-store'; + +export const dynamic = 'force-dynamic'; + +/** + * Readiness probe (#1066), separate from the liveness probe /api/health. + * + * 200 when this instance can serve: always once booted with local files; + * with a shared store, once boot finished and the store answered a ping + * within a second. 503 otherwise, with the reason in `state`. + * + * Kubernetes: readinessProbe here, livenessProbe on /api/health. A store + * outage then takes pods out of the Service without restarting them. + * + * The ping result is reused for a second, so probes can't flood the store. + */ + +const CACHE_MS = 1000; +let last: { at: number; ok: boolean } | null = null; + +async function storeAnswers(): Promise { + if (last && Date.now() - last.at < CACHE_MS) return last.ok; + let ok = true; + try { + await getStore().ping(1000); + } catch { + ok = false; + } + last = { at: Date.now(), ok }; + return ok; +} + +async function readiness(): Promise<{ status: number; body: Record }> { + const store = getStore(); + const state = bootState(); + if (state !== 'ready') { + return { status: 503, body: { ready: false, store: store.mode, state } }; + } + if (store.mode === 'local') return { status: 200, body: { ready: true, store: 'local' } }; + if (!(await storeAnswers())) { + return { status: 503, body: { ready: false, store: store.mode, state: 'store-down', replica: replicaId() } }; + } + return { status: 200, body: { ready: true, store: store.mode, replica: replicaId() } }; +} + +export async function GET() { + const { status, body } = await readiness(); + return NextResponse.json(body, { status, headers: { 'Cache-Control': 'no-store' } }); +} + +export async function HEAD() { + const { status } = await readiness(); + return new NextResponse(null, { status, headers: { 'Cache-Control': 'no-store' } }); +} diff --git a/app/api/webdav/route.ts b/app/api/webdav/route.ts index c9a0de7b3..3db45873b 100644 --- a/app/api/webdav/route.ts +++ b/app/api/webdav/route.ts @@ -6,6 +6,7 @@ import { forwardedClientHeaders, loginClientAddress } from '@/lib/auth/client-ad import { fetchJmapServer } from '@/lib/stalwart/server-fetch'; import { DisallowedUrlError } from '@/lib/security/url-guard'; import { UpstreamBodyTooLargeError, readBodyCapped, readTextCapped } from '@/lib/security/capped-body'; +import { bootGateResponse } from '@/lib/shared-store/boot-gate'; /** A PROPFIND listing of a folder with many thousand entries still fits. */ const MAX_PROPFIND_RESPONSE_BYTES = 50 * 1024 * 1024; @@ -59,6 +60,9 @@ function buildDavTargetUrl(baseUrl: string, username: string, rawPath: string, c * Overwrite: WebDAV Overwrite header for MOVE/COPY */ export async function POST(request: NextRequest) { + // The proxy doesn't run for this route, so it applies the boot gate itself. + const notReady = bootGateResponse(); + if (notReady) return notReady; // CSRF gate (GHSA-9mvj-98f5-9q6g): this handler acts with the caller's // session cookie, which SameSite=Lax still sends from a same-site page. const crossOrigin = rejectCrossOriginRequest(request); diff --git a/app/api/wopi/files/[fileId]/contents/route.ts b/app/api/wopi/files/[fileId]/contents/route.ts index d60ddf0ed..f70fca6b9 100644 --- a/app/api/wopi/files/[fileId]/contents/route.ts +++ b/app/api/wopi/files/[fileId]/contents/route.ts @@ -3,6 +3,7 @@ import { logger } from '@/lib/logger'; import { downloadFileBlob, getWopiFileNode, updateFileNodeBlob, uploadFileBlob } from '@/lib/wopi/files'; import { wopiContext } from '@/lib/wopi/request'; import { readBodyCapped } from '@/lib/security/capped-body'; +import { bootGateResponse } from '@/lib/shared-store/boot-gate'; /** * Largest document PutFile takes. proxy.ts doesn't run for this route, so @@ -39,6 +40,9 @@ export async function GET( request: NextRequest, { params }: { params: Promise<{ fileId: string }> }, ) { + // The proxy doesn't run for this route, so it applies the boot gate itself. + const notReady = bootGateResponse(); + if (notReady) return notReady; // The editor fetches this server-to-server and sends no Sec-Fetch headers. // A browser navigating to it, or loading it as a script or style, is not // the editor. @@ -89,6 +93,9 @@ export async function POST( request: NextRequest, { params }: { params: Promise<{ fileId: string }> }, ) { + // The proxy doesn't run for this route, so it applies the boot gate itself. + const notReady = bootGateResponse(); + if (notReady) return notReady; try { const { fileId: documentId } = await params; const auth = await wopiContext(request, documentId); diff --git a/instrumentation.node.ts b/instrumentation.node.ts index 3157b49de..f73112096 100644 --- a/instrumentation.node.ts +++ b/instrumentation.node.ts @@ -5,6 +5,9 @@ import { migrateLegacyAdminLayout, migratePolicyUnifiedMailbox } from "./lib/adm import { detectSetupState } from "./lib/setup/state"; import { ensureSetupToken } from "./lib/setup/token"; import { ensureSessionSecret } from "./lib/auth/session-secret"; +import { bootState, sharedStoreUrl } from "./lib/shared-store"; +import { bootSharedMode } from "./lib/shared-store/boot"; +import { importLocalData } from "./lib/shared-store/import"; const pkg = JSON.parse( readFileSync(`${process.cwd()}/package.json`, "utf-8") @@ -12,23 +15,40 @@ const pkg = JSON.parse( const current: string = pkg.version ?? "0.0.0"; console.info(`Bulwark Webmail v${current}`); +// With REDIS_URL every replica shares one store (#1066). Without it, +// everything below runs exactly as it always has. +const shared = sharedStoreUrl() !== null; + +async function loadAdmin(): Promise { + await configManager.load(); + await initAdminPassword(); +} + +async function bootShared(): Promise { + await bootSharedMode({ version: current, importLocal: importLocalData, load: loadAdmin }); + if (bootState() !== "ready") throw new Error(`shared store boot ended in state ${bootState()}`); +} + // Initialize admin config and password bootstrap. Migration runs first so -// existing v1 layouts are split before anything reads admin.json. +// existing v1 layouts are split before anything reads admin.json (with a +// shared store: before they are imported). migrateLegacyAdminLayout() .then(() => migratePolicyUnifiedMailbox()) - .then(() => configManager.load()) - .then(() => initAdminPassword()) + .then(() => (shared ? bootShared() : loadAdmin())) .then(async () => { console.info("Admin dashboard initialized"); - try { - if (await ensureSessionSecret()) { - console.info( - "Generated a session secret in the admin state dir. Keep that dir on a " + - "persistent volume, or set SESSION_SECRET, so restarts don't sign everyone out.", - ); + // With a shared store, boot already made or loaded the secret. + if (!shared) { + try { + if (await ensureSessionSecret()) { + console.info( + "Generated a session secret in the admin state dir. Keep that dir on a " + + "persistent volume, or set SESSION_SECRET, so restarts don't sign everyone out.", + ); + } + } catch (err) { + console.warn("Failed to generate a session secret:", err instanceof Error ? err.message : err); } - } catch (err) { - console.warn("Failed to generate a session secret:", err instanceof Error ? err.message : err); } // If we're in bootstrap state (no JMAP_SERVER_URL env and no // setupComplete in config.json), generate/refresh the setup token and diff --git a/lib/auth/session-secret.ts b/lib/auth/session-secret.ts index f2c3738a5..cd3b6cee7 100644 --- a/lib/auth/session-secret.ts +++ b/lib/auth/session-secret.ts @@ -3,26 +3,28 @@ import { writeFile } from 'node:fs/promises'; import { configManager } from '@/lib/admin/config-manager'; import { ensureStateDir, getStatePath } from '@/lib/admin/paths'; import { readFileEnv } from '@/lib/read-file-env'; +import { getStore, isSharedMode } from '@/lib/shared-store'; /** Written on first start when no secret is configured (see ensureSessionSecret). */ -const GENERATED_SECRET_FILE = '.session-secret'; +export const GENERATED_SECRET_FILE = '.session-secret'; + +/** Where the generated secret lives in the shared store. */ +export const SHARED_SECRET_KEY = 'session-secret'; /** - * Resolve the session secret from any of the supported sources, in priority - * order: - * 1. SESSION_SECRET env var - * 2. SESSION_SECRET_FILE-pointed file - * 3. Admin override in config.json (set by the setup wizard) - * 4. The secret generated on first start, in the admin state dir - * - * Returns an empty string when nothing is configured. Callers must treat - * empty as "feature disabled" rather than crashing. - * - * The configManager fallback exists so the web installer can persist the - * secret without touching .env files. It only takes effect if the env vars - * aren't set, so existing deployments aren't affected. + * The generated secret as read from the shared store at boot. It never + * changes once created, so every module copy (proxy, routes) reads it from + * here without waiting on the store. */ -export function getSessionSecret(): string { +const SHARED_CACHE = Symbol.for('bulwark.sessionSecret.shared'); +type SecretGlobals = typeof globalThis & { [SHARED_CACHE]?: string }; + +function sharedSecret(): string { + return (globalThis as SecretGlobals)[SHARED_CACHE] ?? ''; +} + +/** A secret the operator set: env, file, or the setup wizard's config. */ +function configuredSessionSecret(): string { const fromEnv = process.env.SESSION_SECRET; if (fromEnv) return fromEnv; @@ -32,11 +34,29 @@ export function getSessionSecret(): string { const fromAdmin = configManager.get('sessionSecret', ''); if (fromAdmin) return fromAdmin; - const fromAdminFile = readFileEnv( - configManager.get('sessionSecretFile', ''), - ); - if (fromAdminFile) return fromAdminFile; + return readFileEnv(configManager.get('sessionSecretFile', '')) || ''; +} +/** + * Resolve the session secret from any of the supported sources, in priority + * order: + * 1. SESSION_SECRET env var + * 2. SESSION_SECRET_FILE-pointed file + * 3. Admin override in config.json (set by the setup wizard) + * 4. The secret generated on first start: in the admin state dir, or with + * a shared store in the store, one for every replica (#1066) + * + * Returns an empty string when nothing is configured. Callers must treat + * empty as "feature disabled" rather than crashing. + * + * The configManager fallback exists so the web installer can persist the + * secret without touching .env files. It only takes effect if the env vars + * aren't set, so existing deployments aren't affected. + */ +export function getSessionSecret(): string { + const configured = configuredSessionSecret(); + if (configured) return configured; + if (isSharedMode()) return sharedSecret(); return readFileEnv(getStatePath(GENERATED_SECRET_FILE)) || ""; } @@ -51,9 +71,14 @@ export function hasSessionSecret(): boolean { * writable and a volume in docker-compose.yml; without a persistent volume a * restart issues a new one and signs everyone out. * + * With a shared store it lives in the store: the first replica creates it + * and every replica caches the stored value, so replicas starting at the + * same moment end up with one secret. + * * Returns true when this call wrote a new secret. */ export async function ensureSessionSecret(): Promise { + if (isSharedMode()) return ensureSharedSessionSecret(); if (hasSessionSecret()) return false; await ensureStateDir(); try { @@ -68,3 +93,33 @@ export async function ensureSessionSecret(): Promise { throw error; } } + +/** + * Shared mode: create the stored secret if there is none (`value`, or a new + * random one), then load whatever the store holds into this process. Also + * called when a configured secret exists, so the stored one (used once the + * configured one is removed) is loaded either way. + */ +export async function ensureSharedSessionSecret(value?: string): Promise { + const store = getStore(); + const created = await store.kv.set(SHARED_SECRET_KEY, value ?? randomBytes(32).toString('base64'), { ifAbsent: true }); + const stored = await store.kv.get(SHARED_SECRET_KEY); + if (!stored) throw new Error('The shared store lost the session secret right after writing it'); + (globalThis as SecretGlobals)[SHARED_CACHE] = stored; + return created; +} + +/** The generated secret of this install's state dir, if there is one (the import). */ +export function localGeneratedSessionSecret(): string { + return readFileEnv(getStatePath(GENERATED_SECRET_FILE)) || ''; +} + +/** Whether the operator configured a secret (env, file or config). */ +export function hasConfiguredSessionSecret(): boolean { + return configuredSessionSecret().length > 0; +} + +/** Forget the cached shared secret (tests). */ +export function resetSharedSessionSecretForTests(): void { + delete (globalThis as SecretGlobals)[SHARED_CACHE]; +} diff --git a/lib/shared-store/__tests__/boot.test.ts b/lib/shared-store/__tests__/boot.test.ts new file mode 100644 index 000000000..d6964070d --- /dev/null +++ b/lib/shared-store/__tests__/boot.test.ts @@ -0,0 +1,210 @@ +/** + * @vitest-environment node + */ +import { mkdtempSync, rmSync, writeFileSync, mkdirSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('@/lib/admin/config-manager', () => ({ + configManager: { get: vi.fn((_key: string, fallback: unknown) => fallback) }, +})); + +import { bootState, getStore, setBootState, setStoreForTests } from '../index'; +import { resetLocalStoreForTests } from '../local'; +import { + INIT_EPOCH, + announceChange, + markEpochsSeen, + onSharedChange, + pollChanges, + resetChangesForTests, +} from '../changes'; +import { bootSharedMode } from '../boot'; +import { + ensureSharedSessionSecret, + getSessionSecret, + resetSharedSessionSecretForTests, + SHARED_SECRET_KEY, +} from '@/lib/auth/session-secret'; +import { fakeSharedStore } from './fake-shared'; + +let stateDir: string; + +beforeEach(() => { + stateDir = mkdtempSync(path.join(tmpdir(), 'bulwark-boot-')); + vi.stubEnv('ADMIN_STATE_DIR', stateDir); + vi.stubEnv('SESSION_SECRET', ''); + vi.stubEnv('SESSION_SECRET_FILE', ''); + resetLocalStoreForTests(); + resetChangesForTests(); + resetSharedSessionSecretForTests(); +}); + +afterEach(() => { + setStoreForTests(null); + resetChangesForTests(); + vi.unstubAllEnvs(); + rmSync(stateDir, { recursive: true, force: true }); +}); + +const noImport = async () => ({ imported: {} }); + +describe('change notices', () => { + it('runs this process\'s listeners at once in local mode', async () => { + const seen: string[] = []; + onSharedChange('config', (origin) => void seen.push(origin)); + await announceChange('config'); + expect(seen).toEqual(['local']); + }); + + it('runs listeners for another replica\'s change on the next poll, once', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + await store.epochs.bump(INIT_EPOCH); + await markEpochsSeen(); + const seen: string[] = []; + onSharedChange('policy', (origin) => void seen.push(origin)); + await store.epochs.bump('policy'); // another replica saved + await pollChanges(); + await pollChanges(); + expect(seen).toEqual(['remote']); + }); + + it('does not reload its own change, but does when another landed in between', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + await store.epochs.bump(INIT_EPOCH); + await markEpochsSeen(); + const seen: string[] = []; + onSharedChange('config', (origin) => void seen.push(origin)); + await announceChange('config'); + await pollChanges(); + expect(seen).toEqual(['local']); + await store.epochs.bump('config'); // other replica + await announceChange('config'); // ours right after + await pollChanges(); + expect(seen).toEqual(['local', 'local', 'remote']); + }); + + it('retries a listener that failed on the next poll', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + await store.epochs.bump(INIT_EPOCH); + await markEpochsSeen(); + let calls = 0; + onSharedChange('plugins', () => { + calls++; + if (calls === 1) throw new Error('store hiccup'); + }); + await store.epochs.bump('plugins'); + await pollChanges(); + await pollChanges(); + await pollChanges(); + expect(calls).toBe(2); + }); + + it('goes to store-lost when the store forgets the deployment, and back when it returns', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + setBootState('ready'); + await pollChanges(); // no _init epoch: an emptied store + expect(bootState()).toBe('store-lost'); + const reloaded: string[] = []; + onSharedChange('config', (origin) => void reloaded.push(origin)); + await store.epochs.bump(INIT_EPOCH); // restored + await pollChanges(); + expect(bootState()).toBe('ready'); + expect(reloaded).toEqual(['remote']); + }); +}); + +describe('session secret with a shared store', () => { + it('gives every replica the same generated secret', async () => { + setStoreForTests(fakeSharedStore()); + const created = await Promise.all([ensureSharedSessionSecret(), ensureSharedSessionSecret(), ensureSharedSessionSecret()]); + expect(created.filter(Boolean)).toHaveLength(1); + const stored = await getStore().kv.get(SHARED_SECRET_KEY); + expect(stored).toMatch(/^[A-Za-z0-9+/]{43}=$/); + expect(getSessionSecret()).toBe(stored); + }); + + it('ignores the state dir file in shared mode and lets SESSION_SECRET win', async () => { + writeFileSync(path.join(stateDir, '.session-secret'), 'local-file-secret-local-file-secret-xx'); + setStoreForTests(fakeSharedStore()); + await ensureSharedSessionSecret(); + expect(getSessionSecret()).not.toBe('local-file-secret-local-file-secret-xx'); + vi.stubEnv('SESSION_SECRET', 'from-env-from-env-from-env-from-env-xx'); + expect(getSessionSecret()).toBe('from-env-from-env-from-env-from-env-xx'); + }); +}); + +describe('bootSharedMode', () => { + it('initialises an empty store: secret, marker, then loads and becomes ready', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + const order: string[] = []; + await bootSharedMode({ + version: '9.9.9', + importLocal: async () => { order.push('import'); return { imported: { settings: 2 } }; }, + load: async () => { order.push('load'); }, + }); + expect(order).toEqual(['import', 'load']); + expect(bootState()).toBe('ready'); + expect((await store.epochs.all())[INIT_EPOCH]).toBe(1); + expect(JSON.parse((await store.kv.get('meta:initialized'))!)).toMatchObject({ imported: { settings: 2 } }); + expect(getSessionSecret()).toBe(await store.kv.get(SHARED_SECRET_KEY)); + expect((await store.replicas.recent(60_000)).map((r) => r.version)).toEqual(['9.9.9']); + }); + + it('imports this install\'s generated secret before creating one', async () => { + writeFileSync(path.join(stateDir, '.session-secret'), 'old-install-secret-old-install-secret'); + const store = fakeSharedStore(); + setStoreForTests(store); + await bootSharedMode({ version: '1', importLocal: noImport, load: async () => {} }); + expect(await store.kv.get(SHARED_SECRET_KEY)).toBe('old-install-secret-old-install-secret'); + expect(getSessionSecret()).toBe('old-install-secret-old-install-secret'); + }); + + it('joins an initialised store without importing again', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + await bootSharedMode({ version: '1', importLocal: noImport, load: async () => {} }); + const importLocal = vi.fn(noImport); + resetSharedSessionSecretForTests(); + await bootSharedMode({ version: '1', importLocal, load: async () => {} }); + expect(importLocal).not.toHaveBeenCalled(); + expect(bootState()).toBe('ready'); + expect((await store.epochs.all())[INIT_EPOCH]).toBe(1); + }); + + it('refuses to serve when its own local data was encrypted with another secret', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + await bootSharedMode({ version: '1', importLocal: noImport, load: async () => {} }); + mkdirSync(stateDir, { recursive: true }); + writeFileSync(path.join(stateDir, '.session-secret'), 'a-different-install-secret-a-different'); + const load = vi.fn(async () => {}); + await bootSharedMode({ version: '1', importLocal: noImport, load }); + expect(bootState()).toBe('blocked'); + expect(load).not.toHaveBeenCalled(); + }); + + it('waits for a store that is down at start', async () => { + vi.useFakeTimers({ toFake: ['setTimeout', 'setInterval'] }); + try { + const store = fakeSharedStore(); + store.down = true; + setStoreForTests(store); + const done = bootSharedMode({ version: '1', importLocal: noImport, load: async () => {} }); + await vi.advanceTimersByTimeAsync(5000); + expect(bootState()).toBe('starting'); + store.down = false; + await vi.advanceTimersByTimeAsync(5000); + await done; + expect(bootState()).toBe('ready'); + } finally { + vi.useRealTimers(); + } + }); +}); diff --git a/lib/shared-store/__tests__/fake-shared.ts b/lib/shared-store/__tests__/fake-shared.ts new file mode 100644 index 000000000..13e8fd6c0 --- /dev/null +++ b/lib/shared-store/__tests__/fake-shared.ts @@ -0,0 +1,27 @@ +import { createLocalStore } from '../local'; +import type { SharedStore } from '../types'; + +/** + * The local store under the shared store's name: everything that branches + * on isSharedMode() takes its shared path, in one process. The contract + * suite proves both implementations behave alike, so this stands in for a + * Redis in unit tests. `down` makes every call fail as an outage would. + */ +export function fakeSharedStore(): SharedStore & { down: boolean } { + const local = createLocalStore(); + const fake = { + ...local, + mode: 'redis' as const, + down: false, + async ping() { + if (fake.down) { + const { StoreUnavailableError } = await import('../types'); + throw new StoreUnavailableError('down (test)'); + } + }, + status() { + return { mode: 'redis' as const, location: 'redis://fake:6379/0', connected: !fake.down }; + }, + }; + return fake; +} diff --git a/lib/shared-store/__tests__/ready-and-gate.test.ts b/lib/shared-store/__tests__/ready-and-gate.test.ts new file mode 100644 index 000000000..6341107f1 --- /dev/null +++ b/lib/shared-store/__tests__/ready-and-gate.test.ts @@ -0,0 +1,92 @@ +/** + * @vitest-environment node + */ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { NextRequest } from 'next/server'; + +vi.mock('@/lib/admin/config-manager', () => ({ + configManager: { + ensureLoaded: vi.fn(async () => {}), + get: vi.fn((_key: string, fallback: unknown) => fallback), + getPolicy: vi.fn(() => ({ features: { pluginsEnabled: false } })), + }, +})); +vi.mock('@/lib/setup/state', () => ({ detectSetupState: vi.fn(() => 'configured') })); +vi.mock('@/lib/admin/csp-frame-origins', () => ({ getEnabledPluginFrameOrigins: vi.fn(async () => []) })); +vi.mock('next-intl/middleware', async () => { + const { NextResponse } = await import('next/server'); + return { default: () => () => NextResponse.next() }; +}); + +import { proxy } from '@/proxy'; +import { GET as ready } from '@/app/api/ready/route'; +import { configManager } from '@/lib/admin/config-manager'; +import { setBootState, setStoreForTests } from '../index'; +import { fakeSharedStore } from './fake-shared'; + +afterEach(() => { + setStoreForTests(null); + vi.clearAllMocks(); +}); + +function request(path: string): NextRequest { + return new NextRequest('http://localhost:3000' + path, { + headers: { 'sec-fetch-dest': 'document', 'sec-fetch-mode': 'navigate', 'sec-fetch-site': 'none' }, + }); +} + +describe('/api/ready', () => { + it('is ready with local files, as a single instance always was', async () => { + const res = await ready(); + expect(res.status).toBe(200); + expect(await res.json()).toMatchObject({ ready: true, store: 'local' }); + }); + + it('is not ready while a shared-store replica is starting', async () => { + setStoreForTests(fakeSharedStore()); + const res = await ready(); + expect(res.status).toBe(503); + expect(await res.json()).toMatchObject({ ready: false, state: 'starting' }); + }); + + it('follows the store once booted', async () => { + const store = fakeSharedStore(); + setStoreForTests(store); + setBootState('ready'); + expect((await ready()).status).toBe(200); + store.down = true; + await new Promise((r) => setTimeout(r, 1100)); // past the 1 s ping cache + const down = await ready(); + expect(down.status).toBe(503); + expect(await down.json()).toMatchObject({ state: 'store-down' }); + }); +}); + +describe('proxy boot gate', () => { + it('serves nothing but the probes while a shared-store replica is starting', async () => { + setStoreForTests(fakeSharedStore()); + const page = await proxy(request('/en/mail')); + expect(page.status).toBe(503); + expect(page.headers.get('retry-after')).toBe('5'); + const api = await proxy(request('/api/settings')); + expect(api.status).toBe(503); + expect(await api.json()).toMatchObject({ error: 'not_ready', state: 'starting' }); + // The probes pass through to their routes, without touching config. + for (const probe of ['/api/health', '/api/ready']) { + const res = await proxy(request(probe)); + expect(res.status).toBe(200); + expect(res.headers.get('x-middleware-next')).toBe('1'); + } + expect(configManager.ensureLoaded).not.toHaveBeenCalled(); + }); + + it('lets everything through once booted, and always in local mode', async () => { + const page = await proxy(request('/en/mail')); + expect(page.status).not.toBe(503); + setStoreForTests(fakeSharedStore()); + setBootState('ready'); + expect((await proxy(request('/en/mail'))).status).not.toBe(503); + setBootState('store-lost'); + expect((await proxy(request('/en/mail'))).status).toBe(503); + }); +}); diff --git a/lib/shared-store/boot-gate.ts b/lib/shared-store/boot-gate.ts new file mode 100644 index 000000000..1c993d17e --- /dev/null +++ b/lib/shared-store/boot-gate.ts @@ -0,0 +1,25 @@ +import { NextResponse } from 'next/server'; +import { bootState } from './index'; + +const MESSAGES = { + starting: 'Bulwark is starting and connecting to its shared store.', + 'store-lost': 'Bulwark lost its shared store data and is waiting for it to be restored.', + blocked: 'This Bulwark instance has local data that conflicts with the shared store. See the server log.', +} as const; + +/** + * 503 for every request while this replica can't serve (shared mode only: + * local mode is always ready). The proxy calls it for everything but the + * health and readiness probes; the three upload routes the proxy skips call + * it themselves. + */ +export function bootGateResponse(pathname?: string): NextResponse | null { + const state = bootState(); + if (state === 'ready') return null; + const message = MESSAGES[state]; + const headers = { 'Retry-After': '5', 'Cache-Control': 'no-store' }; + if (!pathname || pathname.startsWith('/api/')) { + return NextResponse.json({ error: 'not_ready', state, message }, { status: 503, headers }); + } + return new NextResponse(message, { status: 503, headers: { ...headers, 'Content-Type': 'text/plain; charset=utf-8' } }); +} diff --git a/lib/shared-store/boot.ts b/lib/shared-store/boot.ts new file mode 100644 index 000000000..7a434f56b --- /dev/null +++ b/lib/shared-store/boot.ts @@ -0,0 +1,170 @@ +import { logger } from '@/lib/logger'; +import { + ensureSharedSessionSecret, + hasConfiguredSessionSecret, + localGeneratedSessionSecret, + SHARED_SECRET_KEY, +} from '@/lib/auth/session-secret'; +import { INIT_EPOCH, markEpochsSeen, startChangePoller } from './changes'; +import { getStore, replicaId, setBootState } from './index'; + +/** + * Boot with a shared store (#1066). Order: + * + * 1. Wait until the store answers (the replica reports "starting" meanwhile; + * the proxy answers 503 to everything but the probes). + * 2. Under the `boot` lease, so replicas starting together take turns: + * - Store not initialised yet: import this replica's local files (an + * existing single-instance install moving over), create the session + * secret if none was imported, then mark the store initialised. The + * marker is written last, so an import that dies halfway is redone. + * - Already initialised: check that this replica's local data doesn't + * conflict with it (a generated secret of its own would mean data + * encrypted with another key). + * 3. Load the secret, take the epochs as the baseline, run `load` (config + * and the rest), start the change poller and the replica heartbeat. + */ + +export interface ImportSummary { + /** What was written per store, e.g. { settings: 3, config: 1 }. */ + imported: Record; +} + +export interface SharedBootOptions { + version: string; + /** Copy this replica's local files into an uninitialised store. */ + importLocal: () => Promise; + /** Load everything this process caches from the store (config, …). */ + load: () => Promise; +} + +const BOOT_LEASE = 'boot'; +const BOOT_LEASE_TTL_MS = 60_000; +const HEARTBEAT_MS = 15_000; +const META_KEY = 'meta:initialized'; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function waitForStore(): Promise { + const store = getStore(); + let warned = false; + for (;;) { + try { + await store.ping(2000); + return; + } catch (error) { + if (!warned) { + warned = true; + logger.warn('Waiting for the shared store', { + location: store.status().location, + error: error instanceof Error ? error.message : 'Unknown error', + }); + } + await sleep(2000); + } + } +} + +/** Hold the boot lease while `task` runs, renewing it; wait for it first. */ +async function withBootLease(task: () => Promise): Promise { + const store = getStore(); + const holder = replicaId(); + while (!(await store.leases.acquire(BOOT_LEASE, holder, BOOT_LEASE_TTL_MS))) await sleep(1000); + const renew = setInterval(() => { + void store.leases.acquire(BOOT_LEASE, holder, BOOT_LEASE_TTL_MS).catch(() => {}); + }, BOOT_LEASE_TTL_MS / 3); + renew.unref(); + try { + return await task(); + } finally { + clearInterval(renew); + await store.leases.release(BOOT_LEASE, holder).catch(() => {}); + } +} + +async function initialise(options: SharedBootOptions): Promise<'initialised' | 'joined' | 'blocked'> { + const store = getStore(); + return withBootLease(async () => { + const epochs = await store.epochs.all(); + if (!epochs[INIT_EPOCH]) { + // The local generated secret first: settings and plugin secrets + // imported with it are encrypted with it. + const localSecret = localGeneratedSessionSecret(); + if (localSecret) await store.kv.set(SHARED_SECRET_KEY, localSecret, { ifAbsent: true }); + const summary = await options.importLocal(); + await ensureSharedSessionSecret(); + await store.kv.set(META_KEY, JSON.stringify({ at: new Date().toISOString(), by: replicaId(), ...summary })); + await store.epochs.bump(INIT_EPOCH); + const total = Object.values(summary.imported).reduce((a, b) => a + b, 0); + logger.info( + total > 0 ? 'Shared store initialised from this instance\'s local data' : 'Shared store initialised (no local data to import)', + summary.imported, + ); + return 'initialised'; + } + + // Joining: a generated secret in this replica's state dir that isn't the + // store's means its local data belongs to another install. Serving would + // be fine, but the admin most likely meant to import that data. + const localSecret = localGeneratedSessionSecret(); + const storedSecret = await store.kv.get(SHARED_SECRET_KEY); + if (localSecret && storedSecret && localSecret !== storedSecret && !hasConfiguredSessionSecret()) { + logger.error( + 'This instance has local data (a generated session secret in its state dir) from a single-instance ' + + 'install, but the shared store was already initialised without it. Its settings and plugin ' + + 'credentials could not be read. Either remove the local data dirs from this instance, or empty the ' + + 'shared store and start this instance first so it imports them. See the docs, "Running several instances".', + ); + return 'blocked'; + } + return 'joined'; + }); +} + +function startHeartbeat(version: string): void { + const store = getStore(); + const info = { id: replicaId(), version, startedAt: Date.now() }; + const beat = () => void store.replicas.beat(info).catch(() => {}); + beat(); + setInterval(beat, HEARTBEAT_MS).unref(); +} + +export async function bootSharedMode(options: SharedBootOptions): Promise { + setBootState('starting'); + const store = getStore(); + logger.info('Using a shared store', { location: store.status().location, replica: replicaId() }); + await waitForStore(); + // A store that drops mid-boot is waited for again at each step. + for (;;) { + try { + const result = await initialise(options); + if (result === 'blocked') { + setBootState('blocked'); + return; + } + await ensureSharedSessionSecret(); + if (!hasConfiguredSessionSecret()) { + logger.warn( + 'No SESSION_SECRET set; using the one in the shared store. Set SESSION_SECRET on every replica, ' + + 'so a lost store does not sign everyone out and make stored settings unreadable.', + ); + } + // Baseline before loading, so a change during the load is seen. + await markEpochsSeen(); + await options.load(); + break; + } catch (error) { + logger.warn('Shared store boot step failed; retrying', { + error: error instanceof Error ? error.message : 'Unknown error', + }); + await sleep(2000); + await waitForStore(); + } + } + startChangePoller(); + startHeartbeat(options.version); + setBootState('ready'); + logger.info('Shared store ready', { replica: replicaId() }); +} diff --git a/lib/shared-store/changes.ts b/lib/shared-store/changes.ts new file mode 100644 index 000000000..060e8934f --- /dev/null +++ b/lib/shared-store/changes.ts @@ -0,0 +1,182 @@ +import { logger } from '@/lib/logger'; +import { getStore, isSharedMode, setBootState } from './index'; + +/** + * How admin changes reach every replica (#1066). + * + * Every write to shared state calls announceChange(area): that bumps the + * area's epoch in the store and runs this process's listeners at once (they + * may live in another module copy: proxy, instrumentation, routes). Every + * replica reads all epochs once a second and runs the listeners of each + * area whose epoch moved. A replica therefore sees a change within about a + * second, and one that couldn't reach the store for a while catches up on + * its first successful read. + * + * The epoch is noted before the listeners reload, so a change that lands + * while they read is picked up by the next poll. + * + * Local mode: no poller; announceChange only runs this process's listeners. + */ + +export type ChangeOrigin = 'local' | 'remote'; +export type ChangeListener = (origin: ChangeOrigin) => void | Promise; + +/** Set once the store holds a deployment; its absence later means the store lost its data. */ +export const INIT_EPOCH = '_init'; + +export const POLL_INTERVAL_MS = 1000; + +interface ChangeGlobals { + listeners: Map>; + seen: Map; + poller?: NodeJS.Timeout; + polling?: Promise; + lastPollOk: number; + storeLost: boolean; +} + +const KEY = Symbol.for('bulwark.sharedStore.changes'); + +function globals(): ChangeGlobals { + const g = globalThis as typeof globalThis & { [KEY]?: ChangeGlobals }; + return (g[KEY] ??= { listeners: new Map(), seen: new Map(), lastPollOk: 0, storeLost: false }); +} + +/** + * Run `fn` when `area` changes: at once for a write in this process + * (origin 'local'), and after another replica's write (origin 'remote', + * shared mode only). Returns an unsubscribe function. + */ +export function onSharedChange(area: string, fn: ChangeListener): () => void { + const listeners = globals().listeners; + let set = listeners.get(area); + if (!set) listeners.set(area, (set = new Set())); + set.add(fn); + return () => set!.delete(fn); +} + +async function runListeners(area: string, origin: ChangeOrigin): Promise { + let ok = true; + for (const fn of [...(globals().listeners.get(area) ?? [])]) { + try { + await fn(origin); + } catch (error) { + ok = false; + logger.warn('Could not apply a change from the shared store', { + area, + error: error instanceof Error ? error.message : 'Unknown error', + }); + } + } + return ok; +} + +/** + * Tell every replica that `area` changed. Call it after the write went + * through. In shared mode a failure to bump the epoch is logged, not thrown: + * the write itself succeeded, and other replicas then see it on their next + * reload of that area (or after their next restart) instead. + */ +export async function announceChange(area: string): Promise { + const g = globals(); + if (isSharedMode()) { + try { + const epoch = await getStore().epochs.bump(area); + // Only skip our own change when nothing else landed in between; + // otherwise the poller reloads to pick up the other replica's. + if ((g.seen.get(area) ?? 0) === epoch - 1) g.seen.set(area, epoch); + } catch (error) { + logger.warn('Could not announce a change to the other replicas', { + area, + error: error instanceof Error ? error.message : 'Unknown error', + }); + } + } + await runListeners(area, 'local'); +} + +/** One poll: read the epochs, run the listeners of every area that moved. */ +export async function pollChanges(): Promise { + const g = globals(); + if (g.polling) return g.polling; + g.polling = (async () => { + const epochs = await getStore().epochs.all(); + if (!epochs[INIT_EPOCH]) { + if (!g.storeLost) { + g.storeLost = true; + setBootState('store-lost'); + logger.error( + 'The shared store no longer holds this deployment (restarted without persistence, flushed or evicted). ' + + 'Keeping the last loaded config; nothing is regenerated. Restore the store from a backup, ' + + 'or restart every replica to start it over.', + ); + } + return; + } + g.lastPollOk = Date.now(); + if (g.storeLost) { + // Restored from a backup: reload everything from it. + g.storeLost = false; + g.seen.clear(); + setBootState('ready'); + logger.warn('The shared store holds this deployment again; reloading from it'); + for (const area of g.listeners.keys()) if (!(area in epochs)) await runListeners(area, 'remote'); + } + for (const [area, epoch] of Object.entries(epochs)) { + if (area === INIT_EPOCH) continue; + const before = g.seen.get(area) ?? 0; + if (epoch === before) continue; + g.seen.set(area, epoch); + if (!(await runListeners(area, 'remote'))) { + // Try again on the next poll. + if (g.seen.get(area) === epoch) g.seen.set(area, before); + } + } + })() + .catch(() => { + // Store unreachable: readiness reports it; the next poll catches up. + }) + .finally(() => { + g.polling = undefined; + }); + return g.polling; +} + +/** + * Take the current epochs as the baseline, so the first poll doesn't reload + * what boot just loaded. Call before loading the data the listeners reload. + */ +export async function markEpochsSeen(): Promise { + const epochs = await getStore().epochs.all(); + const g = globals(); + for (const [area, epoch] of Object.entries(epochs)) g.seen.set(area, epoch); +} + +export function startChangePoller(intervalMs = POLL_INTERVAL_MS): void { + const g = globals(); + if (g.poller || !isSharedMode()) return; + g.poller = setInterval(() => void pollChanges(), intervalMs); + g.poller.unref(); +} + +export function stopChangePoller(): void { + const g = globals(); + if (g.poller) clearInterval(g.poller); + g.poller = undefined; +} + +/** When the epochs were last read successfully (ms), 0 if never. */ +export function lastSuccessfulPoll(): number { + return globals().lastPollOk; +} + +export function storeLostItsData(): boolean { + return globals().storeLost; +} + +/** Forget listeners and epochs (tests). */ +export function resetChangesForTests(): void { + stopChangePoller(); + const g = globalThis as typeof globalThis & { [KEY]?: ChangeGlobals }; + delete g[KEY]; +} diff --git a/lib/shared-store/import.ts b/lib/shared-store/import.ts new file mode 100644 index 000000000..0256f5204 --- /dev/null +++ b/lib/shared-store/import.ts @@ -0,0 +1,13 @@ +import type { ImportSummary } from './boot'; + +/** + * Copy an existing single-instance install's local files into an + * uninitialised shared store (#1066). Every store adds its own step here; + * each step writes only what the store doesn't hold yet (put with + * ifVersion: null), so running it again changes nothing, and it never + * changes or deletes a local file. + */ +export async function importLocalData(): Promise { + const imported: Record = {}; + return { imported }; +} diff --git a/proxy.ts b/proxy.ts index 28fd6525f..12f1423ed 100644 --- a/proxy.ts +++ b/proxy.ts @@ -15,6 +15,7 @@ import { listReachableHosts, noThirdPartyCspSources } from "./lib/admin/reachabl import { getWopiEditorOrigins } from "./lib/wopi/discovery"; import { detectSetupState } from "./lib/setup/state"; import { isServedOverHttps } from "./lib/security/request-protocol"; +import { bootGateResponse } from "./lib/shared-store/boot-gate"; const intlMiddleware = createIntlMiddleware(routing); @@ -187,6 +188,19 @@ function isSetupPath(pathname: string): boolean { } export async function proxy(request: NextRequest) { + // The probes answer without config or store, so a replica that is still + // starting (or lost its shared store) can say so (#1066). + const probePath = routePathOf(request.nextUrl.pathname); + if (probePath === "/api/health" || probePath === "/api/ready") { + const probe = NextResponse.next(); + probe.headers.set("X-Content-Type-Options", "nosniff"); + probe.headers.set("X-Frame-Options", "DENY"); + return probe; + } + // With a shared store, nothing else is served until boot finished. + const notReady = bootGateResponse(probePath); + if (notReady) return notReady; + // Resolve setup state before deciding what to skip. The first call after // boot triggers the config load; subsequent calls are in-memory. await configManager.ensureLoaded();