diff --git a/app/api/account-archive/route.ts b/app/api/account-archive/route.ts index 9c77f6ce8..fdbf79c8c 100644 --- a/app/api/account-archive/route.ts +++ b/app/api/account-archive/route.ts @@ -11,6 +11,7 @@ import { rejectCrossOriginRequest } from '@/lib/security/same-origin'; import { MAX_ACCOUNT_SLOTS } from '@/lib/account-utils'; import { archiveIdentity, parseArchiveEnvelope, ARCHIVE_MAX_BYTES, type ArchiveOwner } from '@/lib/account-archive'; import { deleteArchive, loadArchive, saveArchive } from '@/lib/account-archive-storage'; +import { isStoreUnavailable } from '@/lib/stored-data-errors'; export const runtime = 'nodejs'; @@ -18,6 +19,11 @@ const reply = (data: unknown, status = 200) => NextResponse.json(data, { status, headers: { 'Cache-Control': 'no-store' }, }); +/** A shared store that can't be reached is a 503 the client retries; nothing was written. */ +const storageFailed = (error: unknown) => isStoreUnavailable(error) + ? NextResponse.json({ error: 'storage_unavailable' }, { status: 503, headers: { 'Cache-Control': 'no-store', 'Retry-After': '10' } }) + : reply({ error: 'storage_failed' }, 500); + /** * The owner whose archive this request may touch, or the response refusing it. * @@ -67,7 +73,7 @@ export async function GET(request: NextRequest) { const owner = await verifiedOwner(request); if (owner instanceof NextResponse) return owner; try { return reply({ envelope: await loadArchive(owner) }); } - catch { return reply({ error: 'storage_failed' }, 500); } + catch (error) { return storageFailed(error); } } export async function PUT(request: NextRequest) { @@ -83,7 +89,7 @@ export async function PUT(request: NextRequest) { envelope = parseArchiveEnvelope((JSON.parse(text) as { envelope?: unknown })?.envelope); } catch { return reply({ error: 'invalid_archive' }, 400); } try { await saveArchive(owner, envelope); return reply({ ok: true }); } - catch { return reply({ error: 'storage_failed' }, 500); } + catch (error) { return storageFailed(error); } } export async function DELETE(request: NextRequest) { @@ -92,5 +98,5 @@ export async function DELETE(request: NextRequest) { const owner = await verifiedOwner(request); if (owner instanceof NextResponse) return owner; try { await deleteArchive(owner); return reply({ ok: true }); } - catch { return reply({ error: 'storage_failed' }, 500); } + catch (error) { return storageFailed(error); } } diff --git a/app/api/admin/telemetry/route.ts b/app/api/admin/telemetry/route.ts index d0bb0f8c1..aac9d0263 100644 --- a/app/api/admin/telemetry/route.ts +++ b/app/api/admin/telemetry/route.ts @@ -4,8 +4,7 @@ import { auditLog } from '@/lib/admin/audit'; import { logger } from '@/lib/logger'; import { effectiveConsent, - loadState, - saveState, + updateState, buildPayload, sendOnce, reschedule, @@ -13,6 +12,7 @@ import { getLoginCounts, resolveEndpointAllowed, } from '@/lib/telemetry'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; /** * GET /api/admin/telemetry @@ -48,6 +48,8 @@ export async function GET(request: NextRequest) { logger.error('telemetry GET error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err); + if (unavailable) return unavailable; return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } } @@ -90,13 +92,15 @@ export async function POST(request: NextRequest) { if (body.consent !== 'on' && body.consent !== 'off') { return NextResponse.json({ error: 'consent must be "on" or "off"' }, { status: 400 }); } - const state = await loadState(); - const before = state.consent; - state.consent = body.consent; - if (body.consent === 'on' && !state.consentedAt) { - state.consentedAt = new Date().toISOString(); - } - await saveState(state); + const consent = body.consent; + const now = new Date().toISOString(); + let before: string | undefined; + // Against the stored state, so a schedule written meanwhile stays. + await updateState((state) => { + before = state.consent; + state.consent = consent; + if (consent === 'on' && !state.consentedAt) state.consentedAt = now; + }); await reschedule(); await auditLog('telemetry.set-consent', { from: before, to: body.consent }, ip); return NextResponse.json({ ok: true }); @@ -113,11 +117,12 @@ export async function POST(request: NextRequest) { return NextResponse.json({ error: check.reason }, { status: 400 }); } } - const state = await loadState(); - const before = state.endpoint; + let before: string | undefined; // Empty means no collector: heartbeats stop until a URL is set again. - state.endpoint = trimmed; - await saveState(state); + const state = await updateState((current) => { + before = current.endpoint; + current.endpoint = trimmed; + }); await reschedule(); await auditLog('telemetry.set-endpoint', { from: before, to: state.endpoint }, ip); return NextResponse.json({ ok: true, endpoint: state.endpoint }); @@ -138,6 +143,8 @@ export async function POST(request: NextRequest) { logger.error('telemetry POST error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err); + if (unavailable) return unavailable; return NextResponse.json({ error: 'Internal server error' }, { status: 500 }); } } diff --git a/app/api/admin/version/route.ts b/app/api/admin/version/route.ts index 004db83d2..c511a2bd7 100644 --- a/app/api/admin/version/route.ts +++ b/app/api/admin/version/route.ts @@ -10,6 +10,7 @@ import { DEFAULT_VERSION_ENDPOINT, } from '@/lib/version-check'; import { isNoThirdPartyTrafficOn } from '@/lib/admin/third-party-mode'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; /** * GET /api/admin/version @@ -40,6 +41,8 @@ export async function GET(request: NextRequest) { logger.error('version admin GET error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' }); + if (unavailable) return unavailable; return NextResponse.json({ error: 'failed' }, { status: 500 }); } } @@ -64,6 +67,8 @@ export async function POST(req: NextRequest) { logger.error('version admin POST error', { error: err instanceof Error ? err.message : 'unknown', }); + const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' }); + if (unavailable) return unavailable; return NextResponse.json({ error: 'failed' }, { status: 500 }); } } diff --git a/app/api/plugin-secrets/route.ts b/app/api/plugin-secrets/route.ts index e4080fcb5..e235b5a0c 100644 --- a/app/api/plugin-secrets/route.ts +++ b/app/api/plugin-secrets/route.ts @@ -12,6 +12,7 @@ import { putUserCredential, type VaultUser, } from '@/lib/plugin-secrets/vault'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; const PLUGIN_ID_RE = /^[a-z0-9][a-z0-9-]*[a-z0-9]$/; const MAX_SECRET_LENGTH = 4096; @@ -35,7 +36,14 @@ export async function GET(request: NextRequest) { if (pluginId !== undefined && !PLUGIN_ID_RE.test(pluginId)) { return NextResponse.json({ error: 'Invalid plugin id' }, { status: 400 }); } - const credentials = await listUserCredentials(user, pluginId); + let credentials; + try { + credentials = await listUserCredentials(user, pluginId); + } catch (error) { + const unavailable = storedDataUnavailableResponse(error, NO_STORE); + if (unavailable) return unavailable; + throw error; + } // For one plugin, also say which of its admin credentials are set up, and // on which servers: the host page checks the links a plugin shows against // them (the plugin itself only gets yes/no, see proxyAccounts). @@ -93,6 +101,8 @@ export async function POST(request: NextRequest) { }); } catch (error) { if (error instanceof VaultFullError) return NextResponse.json({ error: error.message }, { status: 409 }); + const unavailable = storedDataUnavailableResponse(error, NO_STORE); + if (unavailable) return unavailable; throw error; } return NextResponse.json({ ok: true }, { headers: NO_STORE }); @@ -108,7 +118,13 @@ export async function DELETE(request: NextRequest) { const credentialId = request.nextUrl.searchParams.get('credential') ?? undefined; if (!PLUGIN_ID_RE.test(pluginId)) return bad('Invalid plugin id'); if (credentialId !== undefined && !CREDENTIAL_ID_RE.test(credentialId)) return bad('Invalid credential id'); - await deleteUserCredentials(user, pluginId, credentialId); + try { + await deleteUserCredentials(user, pluginId, credentialId); + } catch (error) { + const unavailable = storedDataUnavailableResponse(error, NO_STORE); + if (unavailable) return unavailable; + throw error; + } return NextResponse.json({ ok: true }, { headers: NO_STORE }); } diff --git a/app/api/settings/route.ts b/app/api/settings/route.ts index 18080c4b5..b7223ddf2 100644 --- a/app/api/settings/route.ts +++ b/app/api/settings/route.ts @@ -5,13 +5,14 @@ import { logger } from '@/lib/logger'; import { decryptSession } from '@/lib/auth/crypto'; import { sessionCookieName } from '@/lib/auth/session-cookie'; import { readStalwartAuthContextFromStore } from '@/lib/stalwart/auth-context'; -import { saveUserSettings, loadUserSettings, deleteUserSettings } from '@/lib/settings-sync'; +import { updateUserSettings, loadUserSettings, deleteUserSettings } from '@/lib/settings-sync'; import { configManager } from '@/lib/admin/config-manager'; import { resolvePolicy } from '@/lib/admin/policy-scopes'; import { hasSessionSecret } from '@/lib/auth/session-secret'; import { MAX_ACCOUNT_SLOTS } from '@/lib/account-utils'; import { mergeStoredTemplates } from '@/lib/template-merge'; import { mergeStoredSubscriptions } from '@/lib/calendar-subscription-sync'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; /** Largest settings blob a client may store (templates and signatures included). */ const SETTINGS_MAX_BYTES = 2 * 1024 * 1024; @@ -141,6 +142,8 @@ export async function GET(request: NextRequest) { const message = error instanceof Error ? error.message : 'Unknown error'; const code = (error as NodeJS.ErrnoException).code; logger.error('Settings load error', { error: message, code }); + const unavailable = storedDataUnavailableResponse(error); + if (unavailable) return unavailable; const classified = classifyError(error); return NextResponse.json({ error: classified.message }, { status: classified.status }); } @@ -210,17 +213,22 @@ export async function POST(request: NextRequest) { // Templates and iCal subscriptions are merged into the stored blob rather // than replaced, so a stale tab or device cannot drop ones another added. - const stored = await loadUserSettings(accountName, storageServerUrl(serverUrl)); - await saveUserSettings( + // In one step against what is stored now: two saves at once (two tabs, + // two replicas) each keep the other's additions. + await updateUserSettings( accountName, storageServerUrl(serverUrl), - mergeStoredSubscriptions(stored, mergeStoredTemplates(stored, filteredSettings)), + (stored) => mergeStoredSubscriptions(stored, mergeStoredTemplates(stored, filteredSettings)), ); return NextResponse.json({ ok: true }); } catch (error) { const message = error instanceof Error ? error.message : 'Unknown error'; const code = (error as NodeJS.ErrnoException).code; logger.error('Settings save error', { error: message, code }); + // Stored settings that don't decrypt, or no store: write nothing, so + // the user's real settings survive and the client keeps its copy. + const unavailable = storedDataUnavailableResponse(error); + if (unavailable) return unavailable; const classified = classifyError(error); return NextResponse.json({ error: classified.message }, { status: classified.status }); } @@ -252,6 +260,8 @@ export async function DELETE(request: NextRequest) { const message = error instanceof Error ? error.message : 'Unknown error'; const code = (error as NodeJS.ErrnoException).code; logger.error('Settings delete error', { error: message, code }); + const unavailable = storedDataUnavailableResponse(error); + if (unavailable) return unavailable; const classified = classifyError(error); return NextResponse.json({ error: classified.message }, { status: classified.status }); } diff --git a/app/api/setup/__tests__/step-session-secret.test.ts b/app/api/setup/__tests__/step-session-secret.test.ts index 93c0cc32e..d4478161a 100644 --- a/app/api/setup/__tests__/step-session-secret.test.ts +++ b/app/api/setup/__tests__/step-session-secret.test.ts @@ -19,6 +19,7 @@ vi.mock('@/lib/telemetry', () => ({ effectiveConsent: async () => ({ source: 'env' }), loadState: async () => ({}), saveState: async () => {}, + updateState: async () => ({}), reschedule: async () => {}, })); diff --git a/app/api/setup/__tests__/step-telemetry-consent.test.ts b/app/api/setup/__tests__/step-telemetry-consent.test.ts new file mode 100644 index 000000000..b49cbdc99 --- /dev/null +++ b/app/api/setup/__tests__/step-telemetry-consent.test.ts @@ -0,0 +1,39 @@ +/** + * @vitest-environment node + */ +import { describe, it, expect, vi } from 'vitest'; +import { NextRequest } from 'next/server'; + +// The wizard's telemetry choice used to load the whole state, change it and +// save it back, putting its copy over anything written meanwhile (the +// scheduler leader storing the next heartbeat, on this or another replica). +// It changes only the consent fields of what is stored now. + +vi.mock('@/lib/setup/state', () => ({ detectSetupState: () => 'bootstrap' })); +vi.mock('@/lib/setup/session', () => ({ authenticateWizardRequest: async () => true })); +vi.mock('@/lib/admin/config-manager', () => ({ + configManager: { ensureLoaded: async () => {}, setAdminConfig: async () => {} }, +})); +const stored = vi.hoisted(() => ({ consent: 'off', consentedAt: null as string | null, nextScheduledAt: 'kept' })); +vi.mock('@/lib/telemetry', () => ({ + effectiveConsent: async () => ({ source: 'file' }), + loadState: async () => { throw new Error('read-modify-write of the whole state'); }, + saveState: async () => { throw new Error('read-modify-write of the whole state'); }, + updateState: async (change: (s: typeof stored) => void) => { change(stored); return stored; }, + reschedule: async () => {}, +})); + +import { POST } from '../step/route'; + +describe('POST /api/setup/step (security, telemetry)', () => { + it('turns consent on against the stored state', async () => { + const res = await POST(new NextRequest('http://localhost/api/setup/step', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ step: 'security', values: { telemetryConsent: 'on' } }), + })); + expect(res.status).toBe(200); + expect(stored).toMatchObject({ consent: 'on', nextScheduledAt: 'kept' }); + expect(stored.consentedAt).not.toBeNull(); + }); +}); diff --git a/app/api/setup/step/route.ts b/app/api/setup/step/route.ts index 17ac852ad..33fa0dccf 100644 --- a/app/api/setup/step/route.ts +++ b/app/api/setup/step/route.ts @@ -4,7 +4,7 @@ import { authenticateWizardRequest } from '@/lib/setup/session'; import { configManager } from '@/lib/admin/config-manager'; import { CONFIG_ENV_MAP } from '@/lib/admin/types'; import { parseJmapServers } from '@/lib/admin/jmap-servers'; -import { effectiveConsent, loadState, saveState, reschedule } from '@/lib/telemetry'; +import { effectiveConsent, updateState, reschedule } from '@/lib/telemetry'; import { logger } from '@/lib/logger'; export const dynamic = 'force-dynamic'; @@ -93,12 +93,12 @@ export async function POST(request: NextRequest) { // A BULWARK_TELEMETRY env var hard-locks the choice; don't fight it. const { source } = await effectiveConsent(); if (source !== 'env') { - const tstate = await loadState(); - tstate.consent = consent; - if (consent === 'on' && !tstate.consentedAt) { - tstate.consentedAt = new Date().toISOString(); - } - await saveState(tstate); + const now = new Date().toISOString(); + // Against the stored state, so a schedule written meanwhile stays. + await updateState((tstate) => { + tstate.consent = consent; + if (consent === 'on' && !tstate.consentedAt) tstate.consentedAt = now; + }); await reschedule(); } } diff --git a/app/api/system/update-status/route.ts b/app/api/system/update-status/route.ts index a55c17d7a..512de8c8a 100644 --- a/app/api/system/update-status/route.ts +++ b/app/api/system/update-status/route.ts @@ -2,6 +2,7 @@ import { NextRequest, NextResponse } from 'next/server'; import { checkOnce, freshStatus, loadState } from '@/lib/version-check'; import { configManager } from '@/lib/admin/config-manager'; import { canSeeInstanceDetails } from '@/lib/auth/instance-details'; +import { storedDataUnavailableResponse } from '@/lib/stored-data-errors'; // Public endpoint that returns the latest cached update status. Fed by the // background scheduler started in instrumentation.node.ts; in production we @@ -13,11 +14,18 @@ import { canSeeInstanceDetails } from '@/lib/auth/instance-details'; // requiring a dev-server restart. The 5s upstream timeout in fetchStatus // caps the worst-case latency added to a dev reload. export async function GET(request: NextRequest) { - if (process.env.NODE_ENV === 'development') { - await checkOnce({ reason: 'dev-reload' }); + let state; + try { + if (process.env.NODE_ENV === 'development') { + await checkOnce({ reason: 'dev-reload' }); + } + state = await loadState(); + } catch (err) { + // The shared store is down: the banner stays as the browser has it. + const unavailable = storedDataUnavailableResponse(err, { 'Cache-Control': 'no-store' }); + if (unavailable) return unavailable; + throw err; } - - const state = await loadState(); // freshStatus drops a status persisted by an earlier build - after an // upgrade it would still advertise the version we just upgraded from (#913). let status = freshStatus(state); diff --git a/instrumentation.node.ts b/instrumentation.node.ts index f73112096..3d5d5d4b5 100644 --- a/instrumentation.node.ts +++ b/instrumentation.node.ts @@ -80,16 +80,19 @@ migrateLegacyAdminLayout() // or with BULWARK_TELEMETRY=on; BULWARK_TELEMETRY=off or an empty // collector URL keeps it from sending. // See https://bulwarkmail.org/docs/legal/privacy/telemetry - const { startScheduler, markProcessStart } = await import("./lib/telemetry"); + const { startScheduler, stopScheduler, markProcessStart } = await import("./lib/telemetry"); markProcessStart(); - await startScheduler(); - }) - .then(async () => { // Hourly check against version.telemetry.bulwarkmail.org. Disable with // BULWARK_UPDATE_CHECK=off or override the endpoint with // BULWARK_UPDATE_CHECK_URL. - const { startScheduler } = await import("./lib/version-check"); - await startScheduler(); + const versionCheck = await import("./lib/version-check"); + // Both run on one replica of the deployment, the scheduler leader + // (#1066); a single instance always leads. + const { startSchedulerLeader } = await import("./lib/scheduler-leader"); + await startSchedulerLeader([ + { name: "telemetry", start: startScheduler, stop: stopScheduler }, + { name: "version-check", start: versionCheck.startScheduler, stop: versionCheck.stopScheduler }, + ]); }) .catch((err) => { console.warn("Admin dashboard init skipped:", err instanceof Error ? err.message : err); diff --git a/lib/__tests__/account-archive-shared-store.test.ts b/lib/__tests__/account-archive-shared-store.test.ts new file mode 100644 index 000000000..7d7c58321 --- /dev/null +++ b/lib/__tests__/account-archive-shared-store.test.ts @@ -0,0 +1,120 @@ +// @vitest-environment node +import { mkdtempSync, readdirSync, readFileSync, rmSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { open } from 'node:fs/promises'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { NextRequest } from 'next/server'; +import { setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// Account archives with a shared store (#1066): one document per owner, +// replaced whole, the same on every replica. + +const mocks = vi.hoisted(() => ({ contexts: new Map(), verify: vi.fn() })); +vi.mock('@/lib/admin/config-manager', () => ({ configManager: { + ensureLoaded: async () => {}, + get: (_key: string, fallback: unknown) => fallback, + getPolicy: () => ({ features: { accountArchiveEnabled: true } }), +} })); +vi.mock('@/lib/auth/session-secret', () => ({ hasSessionSecret: () => true })); +vi.mock('next/headers', () => ({ cookies: async () => ({}) })); +vi.mock('@/lib/stalwart/auth-context', () => ({ + readStalwartAuthContextFromStore: (_: unknown, slot: number) => mocks.contexts.get(slot) ?? null, +})); +vi.mock('@/lib/auth/verify-jmap-auth', async importOriginal => ({ + ...await importOriginal(), verifyJmapAuth: mocks.verify, +})); + +import { deleteArchive, loadArchive, saveArchive } from '@/lib/account-archive-storage'; +import { GET, PUT } from '@/app/api/account-archive/route'; +import type { ArchiveEnvelope } from '@/lib/account-archive'; + +const owner = { username: 'owner@example.com', serverUrl: 'https://mail.example.com' }; +const envelope: ArchiveEnvelope = { version: 1, iterations: 600000, salt: 'A'.repeat(22) + '==', iv: 'A'.repeat(16), ciphertext: 'A'.repeat(43) + '=' }; +const other: ArchiveEnvelope = { ...envelope, ciphertext: 'B'.repeat(43) + '=' }; + +let disks: string; +let store: SeparateDiskStore; + +beforeEach(() => { + disks = mkdtempSync(path.join(tmpdir(), 'bw-archive-replicas-')); + vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'a')); + store = separateDiskStore(); + setStoreForTests(store); + mocks.contexts.clear(); + mocks.contexts.set(0, { username: owner.username, serverUrl: owner.serverUrl, authHeader: 'Basic ' + Buffer.from(`${owner.username}:pw`).toString('base64') }); + mocks.verify.mockReset().mockResolvedValue(owner.serverUrl); +}); + +afterEach(() => { + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + store.cleanup(); + rmSync(disks, { recursive: true, force: true }); +}); + +function request(method: string, body?: unknown) { + return new NextRequest('https://webmail.example.com/api/account-archive', { + method, + headers: { 'x-archive-username': owner.username, 'x-archive-server': owner.serverUrl, 'content-type': 'application/json', 'sec-fetch-site': 'same-origin' }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + }); +} + +describe('account archives with a shared store', () => { + it('returns on replica B what replica A stored, and the latest save wins', async () => { + await saveArchive(owner, envelope); + vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'b')); + expect(await loadArchive(owner)).toEqual(envelope); + + await saveArchive(owner, other); + vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, 'a')); + expect(await loadArchive(owner)).toEqual(other); + + await deleteArchive(owner); + expect(await loadArchive(owner)).toBeNull(); + }); + + it('answers 503 while the store is down and keeps what is stored', async () => { + await saveArchive(owner, envelope); + store.down = true; + expect((await PUT(request('PUT', { envelope: other }))).status).toBe(503); + expect((await GET(request('GET'))).status).toBe(503); + store.down = false; + expect(await loadArchive(owner)).toEqual(envelope); + }); +}); + +describe('account archives on a single instance', () => { + it('flush the new file to disk before it replaces the old one', async () => { + setStoreForTests(null); + const handle = await open(path.join(disks, 'probe'), 'w'); + const proto = Object.getPrototypeOf(handle) as { sync: () => Promise }; + await handle.close(); + const sync = vi.spyOn(proto, 'sync'); + try { + await saveArchive(owner, envelope); + expect(sync).toHaveBeenCalledTimes(1); + expect(await loadArchive(owner)).toEqual(envelope); + } finally { + sync.mockRestore(); + } + }); + + it('keep today\'s file: .json under account-archives, mode 0600', async () => { + setStoreForTests(null); + await saveArchive(owner, envelope); + const dir = path.join(disks, 'a', 'account-archives'); + const files = readdirSync(dir); + expect(files).toHaveLength(1); + expect(JSON.parse(readFileSync(path.join(dir, files[0]), 'utf8'))).toEqual(envelope); + if (process.platform !== 'win32') { + expect(statSync(path.join(dir, files[0])).mode & 0o777).toBe(0o600); + expect(statSync(dir).mode & 0o777).toBe(0o700); + } + expect(files[0]).toMatch(/^[0-9a-f]{64}\.json$/); + }); +}); diff --git a/lib/__tests__/helpers/separate-disk-store.ts b/lib/__tests__/helpers/separate-disk-store.ts new file mode 100644 index 000000000..d86428a98 --- /dev/null +++ b/lib/__tests__/helpers/separate-disk-store.ts @@ -0,0 +1,69 @@ +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { localCollection } from '@/lib/shared-store/local'; +import { fakeSharedStore } from '@/lib/shared-store/__tests__/fake-shared'; +import { + StoreUnavailableError, + type DocCollection, + type KvStore, + type LeaseStore, + type SharedStore, +} from '@/lib/shared-store/types'; + +export type SeparateDiskStore = SharedStore & { down: boolean; root: string; cleanup(): void }; + +/** + * A fake shared store (mode 'redis') whose documents live in a folder of + * their own, not in the dirs a store names for local mode. Every replica has + * its own disk: a store that still wrote its local files would not find on + * "replica B" (another SETTINGS_DATA_DIR, …) what it saved on "replica A". + * `down` makes every document, kv and lease call fail as an outage would. + */ +export function separateDiskStore(): SeparateDiskStore { + const root = mkdtempSync(path.join(tmpdir(), 'bw-shared-store-')); + const base = fakeSharedStore(); + const guard = (run: () => Promise): Promise => + store.down ? Promise.reject(new StoreUnavailableError('down (test)')) : run(); + + const kv: KvStore = { + get: (key) => guard(() => base.kv.get(key)), + set: (key, value, options) => guard(() => base.kv.set(key, value, options)), + del: (key) => guard(() => base.kv.del(key)), + take: (key) => guard(() => base.kv.take(key)), + update: (key, fn, options) => guard(() => base.kv.update(key, fn, options)), + }; + const leases: LeaseStore = { + acquire: (name, holder, ttlMs) => guard(() => base.leases.acquire(name, holder, ttlMs)), + release: (name, holder) => guard(() => base.leases.release(name, holder)), + holder: (name) => guard(() => base.leases.holder(name)), + }; + + const store: SeparateDiskStore = { + ...base, + kv, + leases, + root, + get down() { + return base.down; + }, + set down(value: boolean) { + base.down = value; + }, + collection(name, options): DocCollection { + const inner = localCollection(name, { ...options, local: { ...options.local, dir: () => path.join(root, name) } }); + return { + name, + get: (id) => guard(() => inner.get(id)), + put: (id, data, putOptions) => guard(() => inner.put(id, data, putOptions)), + update: (id, fn, updateOptions) => guard(() => inner.update(id, fn, updateOptions)), + delete: (id, deleteOptions) => guard(() => inner.delete(id, deleteOptions)), + list: (prefix) => guard(() => inner.list(prefix)), + }; + }, + cleanup() { + rmSync(root, { recursive: true, force: true }); + }, + }; + return store; +} diff --git a/lib/__tests__/plugin-vault-shared-store.test.ts b/lib/__tests__/plugin-vault-shared-store.test.ts new file mode 100644 index 000000000..a6171ff64 --- /dev/null +++ b/lib/__tests__/plugin-vault-shared-store.test.ts @@ -0,0 +1,99 @@ +// @vitest-environment node +import { mkdtempSync, readdirSync, rmSync, statSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// Users' plugin credentials with a shared store (#1066). The vault used to +// serialise writes with a queue in its own module, which no other process +// (and no other module copy) sees: two writes on two replicas lost one. + +vi.mock('@/lib/auth/session-secret', () => ({ + getSessionSecret: () => 's'.repeat(64), + hasSessionSecret: () => true, +})); +vi.mock('@/lib/logger', () => ({ + logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} }, +})); + +const user = { username: 'alice@example.org', serverUrl: 'https://mail.example.org' }; +const input = (secret: string) => ({ origin: 'https://cloud.example.org', inject: { type: 'basic' } as const, label: 'Cloud', secret }); + +type Vault = typeof import('@/lib/plugin-secrets/vault'); + +/** A fresh copy of the vault module, as another process (or bundle) has. */ +async function replica(): Promise { + vi.resetModules(); + return import('@/lib/plugin-secrets/vault'); +} + +let disks: string; +let store: SeparateDiskStore; + +beforeEach(() => { + disks = mkdtempSync(path.join(tmpdir(), 'bw-vault-replicas-')); + vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'a')); + store = separateDiskStore(); + setStoreForTests(store); +}); + +afterEach(() => { + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + store.cleanup(); + rmSync(disks, { recursive: true, force: true }); +}); + +describe('plugin credentials with a shared store', () => { + it('reads on replica B what replica A stored', async () => { + const a = await replica(); + await a.putUserCredential(user, 'nextcloud', 'cloud', input('pw')); + + vi.stubEnv('PLUGIN_SECRETS_DIR', path.join(disks, 'b')); + const b = await replica(); + expect(await b.getUserCredential(user, 'nextcloud', 'cloud')).toMatchObject({ secret: 'pw' }); + }); + + it('keeps both of two credentials saved at once through two replicas', async () => { + const a = await replica(); + const b = await replica(); + await Promise.all([ + a.putUserCredential(user, 'nextcloud', 'cloud', input('one')), + b.putUserCredential(user, 'paperless', 'docs', input('two')), + a.putUserCredential(user, 'radicale', 'dav', input('three')), + b.putUserCredential(user, 'immich', 'photos', input('four')), + ]); + + const ids = (await a.listUserCredentials(user)).map((c) => c.pluginId).sort(); + expect(ids).toEqual(['immich', 'nextcloud', 'paperless', 'radicale']); + }); + + it('answers with StoreUnavailableError while the store is down and writes nothing', async () => { + const vault = await replica(); + await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw')); + store.down = true; + await expect(vault.putUserCredential(user, 'paperless', 'docs', input('x'))).rejects.toThrow('down (test)'); + await expect(vault.listUserCredentials(user)).rejects.toThrow('down (test)'); + store.down = false; + expect((await vault.listUserCredentials(user)).map((c) => c.pluginId)).toEqual(['nextcloud']); + }); +}); + +describe('plugin credentials on a single instance', () => { + it('keep today\'s file: .enc in the vault dir, mode 0600, gone when the last one goes', async () => { + setStoreForTests(null); + const vault = await replica(); + await vault.putUserCredential(user, 'nextcloud', 'cloud', input('pw')); + const dir = path.join(disks, 'a'); + const files = readdirSync(dir); + expect(files).toEqual([`${vault.userKey(user)}.enc`]); + if (process.platform !== 'win32') expect(statSync(path.join(dir, files[0])).mode & 0o777).toBe(0o600); + + await vault.deleteUserCredentials(user, 'nextcloud'); + expect(readdirSync(dir)).toEqual([]); + }); +}); diff --git a/lib/__tests__/scheduler-leader.test.ts b/lib/__tests__/scheduler-leader.test.ts new file mode 100644 index 000000000..6b317aa86 --- /dev/null +++ b/lib/__tests__/scheduler-leader.test.ts @@ -0,0 +1,381 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { setTimeout as realSleep } from 'node:timers/promises'; +import { NextRequest } from 'next/server'; +import { replicaId, setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { INIT_EPOCH, announceChange, pollChanges, resetChangesForTests } from '@/lib/shared-store/changes'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// The background jobs with several replicas (#1066): only the holder of the +// scheduler-leader lease runs them, another replica takes over when the +// holder's lease runs out and keeps the stored schedule, and a scheduled +// heartbeat is claimed with a compare-and-set so two leaders can't both send. + +const log = vi.hoisted(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() })); +vi.mock('@/lib/logger', () => ({ logger: log })); +vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false })); +const buildPayload = vi.hoisted(() => vi.fn(async () => ({ schema: '1' }))); +vi.mock('@/lib/telemetry/payload', () => ({ buildPayload, markProcessStart: () => {} })); +const sent = vi.hoisted(() => vi.fn(async (_url: string, _init?: unknown) => new Response('ok'))); +vi.mock('@/lib/telemetry/endpoint-guard', () => ({ + resolveEndpointAllowed: async () => ({ ok: true }), + fetchTelemetryTarget: (url: string, init?: unknown) => sent(url, init), +})); +vi.mock('@/lib/admin/session', () => ({ + requireAdminAuth: async () => ({ session: { sub: 'admin' } }), + getClientIP: () => '127.0.0.1', +})); +vi.mock('@/lib/admin/audit', () => ({ auditLog: async () => {} })); + +import { + LEADER_LEASE, isSchedulerLeader, resetSchedulerLeaderForTests, schedulerLeaderSettled, startSchedulerLeader, stopSchedulerLeader, +} from '@/lib/scheduler-leader'; +import { loadState, saveState, sendOnce, startScheduler, stopScheduler, updateState } from '@/lib/telemetry'; +import { POST as telemetryRoute } from '@/app/api/admin/telemetry/route'; + +const MIN = 60_000; +const COLLECTOR = 'https://collector.example.org/v1/heartbeat'; +const jobs = [{ name: 'telemetry', start: startScheduler, stop: stopScheduler }]; + +let dir: string; +let store: SeparateDiskStore; + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'bw-leader-')); + vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry')); + vi.stubEnv('BULWARK_TELEMETRY', ''); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout', 'setInterval', 'clearInterval', 'Date'] }); + vi.setSystemTime(new Date('2026-10-31T12:00:00.000Z')); + store = separateDiskStore(); + setStoreForTests(store); + sent.mockClear(); + log.info.mockClear(); +}); + +afterEach(async () => { + await stopSchedulerLeader(); + resetSchedulerLeaderForTests(); + await stopScheduler(); + resetChangesForTests(); + vi.useRealTimers(); + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + store.cleanup(); + rmSync(dir, { recursive: true, force: true }); +}); + +async function consentOn(nextScheduledAt: string | null) { + await saveState({ ...(await loadState()), consent: 'on', endpoint: COLLECTOR, nextScheduledAt }); +} + +/** Let time pass; the leader's tries and job starts do real file I/O, so wait for them too. */ +async function pass(ms: number) { + for (let passed = 0; passed < ms; passed += 10_000) { + await vi.advanceTimersByTimeAsync(Math.min(10_000, ms - passed)); + await schedulerLeaderSettled(); + } +} + +/** Another replica holding (and renewing) the lease while `ms` pass here. */ +async function otherReplicaLeads(ms: number) { + for (let passed = 0; passed < ms; passed += 10_000) { + expect(await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000)).toBe(true); + await pass(10_000); + } +} + +/** Let a heartbeat a timer set off finish its real file I/O, so "not sent" means it. */ +async function ioSettles() { + await realSleep(200); +} + +/** The heartbeat the timer set off does real I/O before it sends. */ +async function sends(times: number) { + await vi.waitFor(() => expect(sent).toHaveBeenCalledTimes(times)); +} + +describe('scheduler leader with a shared store', () => { + it('only the holder sends; when its lease runs out another replica takes over and keeps the schedule', async () => { + const slot = new Date(Date.now() + 20 * MIN).toISOString(); + await consentOn(slot); + await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000); + + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(false); + await otherReplicaLeads(5 * MIN); + await ioSettles(); + expect(isSchedulerLeader()).toBe(false); + + // replica-a is gone: its lease expires after 30 s, the next try takes it. + await pass(40_000); + expect(isSchedulerLeader()).toBe(true); + expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.objectContaining({ replica: expect.any(String) })); + expect((await loadState()).nextScheduledAt).toBe(slot); + expect(sent).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(new Date(slot).getTime() - Date.now() + 1000); + await sends(1); + }); + + it('stops its timers when it can no longer reach the store', async () => { + const slot = new Date(Date.now() + 5 * MIN).toISOString(); + await consentOn(slot); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + store.down = true; + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: expect.stringContaining('shared store') })); + // Past the slot with the store still down: no heartbeat from here. + await pass(6 * MIN); + await ioSettles(); + expect(sent).not.toHaveBeenCalled(); + expect(isSchedulerLeader()).toBe(false); + }); + + it('stops leading when the store lost its data', async () => { + await store.epochs.bump(INIT_EPOCH); + await consentOn(new Date(Date.now() + 5 * MIN).toISOString()); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + // Flushed: the init epoch is gone, and the next poll notices. + resetLocalStoreForTests(); + await pollChanges(); + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + expect(log.info).toHaveBeenCalledWith('scheduler: stopped leading', expect.objectContaining({ reason: 'the shared store lost its data' })); + }); + + it('stops its timers when another replica has taken the lease', async () => { + const slot = new Date(Date.now() + 5 * MIN).toISOString(); + await consentOn(slot); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + + // This replica's lease is gone (it stalled past the TTL) and replica-a took it. + await store.leases.release(LEADER_LEASE, replicaId()); + await otherReplicaLeads(6 * MIN); + await ioSettles(); + expect(isSchedulerLeader()).toBe(false); + expect(sent).not.toHaveBeenCalled(); + }); + + it('does not arm a new timer after a send that was in flight when the lead moved', async () => { + const slot = new Date(Date.now() + MIN).toISOString(); + await consentOn(slot); + let finish!: (res: Response) => void; + sent.mockImplementationOnce(() => new Promise((resolve) => { finish = resolve; })); + await startSchedulerLeader(jobs); + + await vi.advanceTimersByTimeAsync(MIN + 1000); + await sends(1); + // The lead moves to replica-a while the heartbeat hangs. + await store.leases.release(LEADER_LEASE, replicaId()); + expect(await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000)).toBe(true); + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + await stopSchedulerLeader(); // no more tries from this replica + + finish(new Response('ok')); + await ioSettles(); + await vi.advanceTimersByTimeAsync(27 * 60 * MIN); + await ioSettles(); + expect(sent).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + + it('stops a job whose restart finished after the replica stepped down', async () => { + let running = false; + let starts = 0; + let finishStart: (() => void) | null = null; + const job = { + name: 'telemetry', + start: async () => { + if (++starts === 2) await new Promise((resolve) => { finishStart = resolve; }); + running = true; + }, + stop: async () => { running = false; }, + }; + await startSchedulerLeader([job]); + expect(running).toBe(true); + + // Consent changed: the restart's start hangs while the lead is lost. + const restarted = announceChange('telemetry'); + await vi.waitFor(() => expect(finishStart).not.toBeNull()); + store.down = true; + await pass(10_000); + expect(isSchedulerLeader()).toBe(false); + finishStart!(); + await restarted; + expect(running).toBe(false); + }); + + it('version check: does not arm a new timer after a check that was in flight when it stopped', async () => { + vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(dir, 'version-check')); + vi.stubEnv('BULWARK_UPDATE_CHECK', ''); + vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/'); + vi.stubEnv('NEXT_PUBLIC_APP_VERSION', '1.20.0'); + let finish!: (res: Response) => void; + const fetchMock = vi.fn(() => new Promise((resolve) => { finish = resolve; })); + vi.stubGlobal('fetch', fetchMock); + const versionCheck = await import('@/lib/version-check'); + try { + await versionCheck.startScheduler(); + await vi.advanceTimersByTimeAsync(31_000); + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1)); + await versionCheck.stopScheduler(); + finish(new Response('{}', { status: 500 })); + await ioSettles(); + await vi.advanceTimersByTimeAsync(2 * 60 * MIN); + await ioSettles(); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + await versionCheck.stopScheduler(); + vi.unstubAllGlobals(); + } + }); + + it('sends once when two leaders fire for the same slot, and not again after it', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + + const results = await Promise.all([ + sendOnce({ reason: 'scheduled', slot }), + sendOnce({ reason: 'scheduled', slot }), + ]); + expect(results.filter((r) => r.ok)).toHaveLength(1); + expect(sent).toHaveBeenCalledTimes(1); + + // The paused leader wakes up an hour later, still holding that slot. + vi.advanceTimersByTime(60 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('sends once when the old leader stalled past its slot and the new one picked a later slot', async () => { + const oldSlot = new Date(Date.now() - 1000).toISOString(); + await consentOn(oldSlot); + // The new leader found the slot passed and scheduled an hour out. + const newSlot = new Date(Date.now() + 60 * MIN).toISOString(); + // The stalled leader wakes up and sends for its slot first. + expect((await sendOnce({ reason: 'scheduled', slot: oldSlot })).ok).toBe(true); + vi.advanceTimersByTime(61 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot: newSlot })).ok).toBe(false); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('sends once when a leader died after sending but before storing the next slot', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + // nextScheduledAt still names the slot just sent; the next leader + // finds it passed and fires an hour later. + vi.advanceTimersByTime(60 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(false); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('still sends the next scheduled heartbeat after a manual "send now"', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + vi.advanceTimersByTime(20 * 60 * MIN); + expect((await sendOnce({ reason: 'admin-manual' })).ok).toBe(true); + vi.advanceTimersByTime(2 * 60 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true); + expect(sent).toHaveBeenCalledTimes(3); + }); + + it('gives the slot back when the payload cannot be built', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + buildPayload.mockRejectedValueOnce(new Error('store down')); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false); + expect((await loadState()).lastSentAt).toBeNull(); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('gives the slot back when the send fails', async () => { + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + sent.mockResolvedValueOnce(new Response('no', { status: 500 })); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(false); + expect((await loadState()).lastSentAt).toBeNull(); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + }); + + it('sends on "send now" from a replica that does not lead, even right after a heartbeat', async () => { + await consentOn(null); + await store.leases.acquire(LEADER_LEASE, 'replica-a', 30_000); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(false); + await updateState((state) => { state.lastSentAt = new Date().toISOString(); }); + + const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'send-now' }), + })); + expect(res.status).toBe(200); + expect(sent).toHaveBeenCalledTimes(1); + }); + + it('starts the heartbeat on the leader when consent is turned on through another replica', async () => { + await store.epochs.bump(INIT_EPOCH); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + expect((await loadState()).nextScheduledAt).toBeNull(); + + // The other replica's admin route: write, then bump the epoch. + await updateState((state) => { state.consent = 'on'; state.endpoint = COLLECTOR; }); + await store.epochs.bump('telemetry'); + await pollChanges(); + + expect((await loadState()).nextScheduledAt).not.toBeNull(); + await vi.advanceTimersByTimeAsync(61 * MIN); + await sends(1); + }); +}); + +describe('scheduler leader on a single instance', () => { + it('sends a scheduled heartbeat an hour after the last one, as before', async () => { + setStoreForTests(null); + const slot = new Date(Date.now() - 1000).toISOString(); + await consentOn(slot); + expect((await sendOnce({ reason: 'scheduled', slot })).ok).toBe(true); + vi.advanceTimersByTime(60 * MIN); + expect((await sendOnce({ reason: 'scheduled', slot: new Date().toISOString() })).ok).toBe(true); + expect((await loadState()).lastScheduledSentAt).toBeUndefined(); + }); + + it('leads at once and restarts the heartbeat when consent changes in this process', async () => { + setStoreForTests(null); + await startSchedulerLeader(jobs); + expect(isSchedulerLeader()).toBe(true); + expect(log.info).toHaveBeenCalledWith('scheduler: leading', expect.anything()); + + const res = await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'set-consent', consent: 'on' }), + })); + expect(res.status).toBe(200); + expect(vi.getTimerCount()).toBe(1); + + await telemetryRoute(new NextRequest('http://localhost/api/admin/telemetry', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ action: 'set-consent', consent: 'off' }), + })); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/lib/__tests__/scheduler-timer-module-copies.test.ts b/lib/__tests__/scheduler-timer-module-copies.test.ts new file mode 100644 index 000000000..8beb553ee --- /dev/null +++ b/lib/__tests__/scheduler-timer-module-copies.test.ts @@ -0,0 +1,62 @@ +// @vitest-environment node +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +// Next loads lib/telemetry and lib/version-check once per bundle. The timer +// sat in a module-level variable, so the admin route's reschedule() acted on +// its own copy: the timer instrumentation started kept running next to the +// new one (two heartbeats), and turning consent off didn't stop it. + +vi.mock('@/lib/admin/third-party-mode', () => ({ isNoThirdPartyTrafficOn: async () => false })); + +let dir: string; + +beforeEach(() => { + dir = mkdtempSync(path.join(tmpdir(), 'bw-scheduler-copies-')); + vi.stubEnv('TELEMETRY_DATA_DIR', path.join(dir, 'telemetry')); + vi.stubEnv('VERSION_CHECK_DATA_DIR', path.join(dir, 'version-check')); + vi.stubEnv('BULWARK_TELEMETRY', 'on'); + vi.stubEnv('BULWARK_UPDATE_CHECK', ''); + vi.stubEnv('BULWARK_UPDATE_CHECK_URL', 'https://version.example/'); + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); +}); + +afterEach(() => { + vi.useRealTimers(); + vi.unstubAllEnvs(); + vi.resetModules(); + rmSync(dir, { recursive: true, force: true }); +}); + +async function copy(load: () => Promise): Promise { + vi.resetModules(); + return load(); +} + +describe('scheduler timers across module copies', () => { + it('telemetry: a stop from another copy clears the running timer, a restart leaves one', async () => { + const instrumentation = await copy(() => import('@/lib/telemetry/sender')); + const route = await copy(() => import('@/lib/telemetry/sender')); + + await instrumentation.startScheduler(); + expect(vi.getTimerCount()).toBe(1); + await route.startScheduler(); + expect(vi.getTimerCount()).toBe(1); + await route.stopScheduler(); + expect(vi.getTimerCount()).toBe(0); + }); + + it('version check: the same', async () => { + const instrumentation = await copy(() => import('@/lib/version-check/sender')); + const route = await copy(() => import('@/lib/version-check/sender')); + + await instrumentation.startScheduler(); + expect(vi.getTimerCount()).toBe(1); + await route.startScheduler(); + expect(vi.getTimerCount()).toBe(1); + await route.stopScheduler(); + expect(vi.getTimerCount()).toBe(0); + }); +}); diff --git a/lib/__tests__/settings-route-account-key.test.ts b/lib/__tests__/settings-route-account-key.test.ts index e4ded638d..f6f99e7a6 100644 --- a/lib/__tests__/settings-route-account-key.test.ts +++ b/lib/__tests__/settings-route-account-key.test.ts @@ -39,6 +39,9 @@ vi.mock('@/lib/settings-sync', () => ({ saveUserSettings: async (username: string, serverUrl: string, settings: Record) => { store.set(`${username}|${serverUrl}`, settings); }, + updateUserSettings: async (username: string, serverUrl: string, change: (stored: Record | null) => Record) => { + store.set(`${username}|${serverUrl}`, change(store.get(`${username}|${serverUrl}`) ?? null)); + }, deleteUserSettings: async (username: string, serverUrl: string) => { store.delete(`${username}|${serverUrl}`); }, })); diff --git a/lib/__tests__/settings-route-template-merge.test.ts b/lib/__tests__/settings-route-template-merge.test.ts index 662edc1d6..befaf929c 100644 --- a/lib/__tests__/settings-route-template-merge.test.ts +++ b/lib/__tests__/settings-route-template-merge.test.ts @@ -38,6 +38,9 @@ vi.mock('@/lib/settings-sync', () => ({ saveUserSettings: async (username: string, serverUrl: string, settings: Record) => { store.set(`${username}|${serverUrl}`, settings); }, + updateUserSettings: async (username: string, serverUrl: string, change: (stored: Record | null) => Record) => { + store.set(`${username}|${serverUrl}`, change(store.get(`${username}|${serverUrl}`) ?? null)); + }, deleteUserSettings: async () => {}, })); diff --git a/lib/__tests__/settings-shared-store.test.ts b/lib/__tests__/settings-shared-store.test.ts new file mode 100644 index 000000000..3716f3a60 --- /dev/null +++ b/lib/__tests__/settings-shared-store.test.ts @@ -0,0 +1,176 @@ +// @vitest-environment node +import { mkdtempSync, readdirSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { NextRequest } from 'next/server'; +import { setStoreForTests } from '@/lib/shared-store'; +import { resetLocalStoreForTests } from '@/lib/shared-store/local'; +import { SETTINGS_COLLECTION, settingsId } from '@/lib/settings-sync'; +import { separateDiskStore, type SeparateDiskStore } from './helpers/separate-disk-store'; + +// Synced settings with a shared store (#1066): every replica reads and +// writes the one store, never its own disk. + +vi.mock('@/lib/auth/session-secret', () => ({ + getSessionSecret: () => 's'.repeat(64), + hasSessionSecret: () => true, +})); +vi.mock('@/lib/logger', () => ({ + logger: { warn: () => {}, error: () => {}, info: () => {}, debug: () => {} }, +})); +vi.mock('@/lib/admin/config-manager', () => ({ + configManager: { + ensureLoaded: async () => {}, + get: (key: string, fallback: unknown) => (key === 'settingsSyncEnabled' ? true : fallback), + getPolicy: () => ({ restrictions: {} }), + }, +})); + +const jar = new Map(); +vi.mock('next/headers', () => ({ + cookies: async () => ({ + get: (name: string) => (jar.has(name) ? { name, value: jar.get(name)! } : undefined), + set: (name: string, value: string) => { jar.set(name, value); }, + delete: (name: string) => { jar.delete(name); }, + getAll: () => [...jar].map(([name, value]) => ({ name, value })), + }), +})); + +const SERVER = 'https://mail.example.org'; +const USER = 'alice@example.org'; + +let disks: string; +let store: SeparateDiskStore; + +/** Point this process at one replica's own data dir. */ +function onReplica(name: 'a' | 'b') { + vi.stubEnv('SETTINGS_DATA_DIR', path.join(disks, name, 'settings')); +} + +beforeEach(async () => { + disks = mkdtempSync(path.join(tmpdir(), 'bw-settings-replicas-')); + store = separateDiskStore(); + setStoreForTests(store); + vi.stubEnv('SETTINGS_SYNC_ENABLED', 'true'); + onReplica('a'); + jar.clear(); + const { setStalwartAuthContextInStore } = await import('@/lib/stalwart/auth-context'); + setStalwartAuthContextInStore({ + get: (n: string) => (jar.has(n) ? { name: n, value: jar.get(n)! } : undefined), + set: (n: string, v: string) => { jar.set(n, v); }, + delete: (n: string) => { jar.delete(n); }, + } as never, 0, { serverUrl: SERVER, authHeader: 'Bearer t', username: USER }); +}); + +afterEach(() => { + setStoreForTests(null); + resetLocalStoreForTests(); + vi.unstubAllEnvs(); + store.cleanup(); + rmSync(disks, { recursive: true, force: true }); +}); + +function template(id: string) { + const at = '2026-10-01T00:00:00.000Z'; + return { id, name: id, subject: '', body: '', isHTML: false, category: '', isFavorite: false, createdAt: at, updatedAt: at }; +} + +async function push(settings: Record) { + const { POST } = await import('@/app/api/settings/route'); + return POST(new NextRequest('https://webmail.example/api/settings', { + method: 'POST', + headers: { 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' }, + body: JSON.stringify({ username: USER, serverUrl: SERVER, settings }), + })); +} + +async function fetchSettings() { + const { GET } = await import('@/app/api/settings/route'); + return GET(new NextRequest('https://webmail.example/api/settings', { + headers: { 'x-settings-username': USER, 'x-settings-server': SERVER }, + })); +} + +function storedDoc() { + return store.collection('settings', SETTINGS_COLLECTION).get(settingsId(USER, SERVER)); +} + +describe('settings with a shared store', () => { + it('loads on replica B what was saved through replica A', async () => { + expect((await push({ theme: 'dark' })).status).toBe(200); + + onReplica('b'); + const res = await fetchSettings(); + expect(res.status).toBe(200); + expect((await res.json()).settings).toEqual({ theme: 'dark' }); + }); + + it('deletes them for every replica', async () => { + await push({ theme: 'dark' }); + onReplica('b'); + const { DELETE } = await import('@/app/api/settings/route'); + const res = await DELETE(new NextRequest('https://webmail.example/api/settings', { + method: 'DELETE', + headers: { 'sec-fetch-site': 'same-origin', 'content-type': 'application/json' }, + body: JSON.stringify({ username: USER, serverUrl: SERVER }), + })); + expect(res.status).toBe(200); + onReplica('a'); + expect((await (await fetchSettings()).json()).settings).toBeNull(); + }); + + it('moves a stored document that does not decrypt aside in the store before a save, keeping its bytes', async () => { + const garbage = Buffer.alloc(64, 7); + const id = settingsId(USER, SERVER); + await store.collection('settings', SETTINGS_COLLECTION).put(id, garbage); + + expect((await (await fetchSettings()).json()).settings).toBeNull(); + expect((await push({ theme: 'light' })).status).toBe(200); + expect((await (await fetchSettings()).json()).settings).toEqual({ theme: 'light' }); + + // In the store, not on this replica's disk. + const aside = readdirSync(path.join(store.root, 'settings')).filter((f) => f.startsWith(`${id}.unreadable-`)); + expect(aside).toHaveLength(1); + const asideId = aside[0].slice(0, -'.enc'.length); + expect((await store.collection('settings', SETTINGS_COLLECTION).get(asideId))?.data.equals(garbage)).toBe(true); + }); + + it('answers 503 and writes nothing while the store is down', async () => { + await push({ theme: 'dark' }); + const before = (await storedDoc())!.data; + + store.down = true; + expect((await push({ theme: 'light' })).status).toBe(503); + expect((await fetchSettings()).status).toBe(503); + store.down = false; + + expect((await storedDoc())!.data.equals(before)).toBe(true); + }); +}); + +// Two saves at once, each from a tab that added a template the other +// doesn't know yet. The route used to load, merge and save without a lock, +// so the second save put its merge over the first: one template was lost, +// with two tabs on one instance as well as across replicas. +async function twoConcurrentSaves() { + await push({ templates: [], deletedTemplateIds: {} }); + const [a, b] = await Promise.all([ + push({ templates: [template('from-a')], deletedTemplateIds: {} }), + push({ templates: [template('from-b')], deletedTemplateIds: {} }), + ]); + expect([a.status, b.status]).toEqual([200, 200]); + const saved = (await (await fetchSettings()).json()).settings as { templates: { id: string }[] }; + return saved.templates.map((t) => t.id).sort(); +} + +describe('two saves at once', () => { + it('keep both templates with a shared store', async () => { + expect(await twoConcurrentSaves()).toEqual(['from-a', 'from-b']); + }); + + it('keep both templates on a single instance', async () => { + setStoreForTests(null); + expect(await twoConcurrentSaves()).toEqual(['from-a', 'from-b']); + }); +}); diff --git a/lib/__tests__/stored-data-unreadable.test.ts b/lib/__tests__/stored-data-unreadable.test.ts new file mode 100644 index 000000000..d709f8366 --- /dev/null +++ b/lib/__tests__/stored-data-unreadable.test.ts @@ -0,0 +1,212 @@ +// @vitest-environment node +import { mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { NextRequest } from 'next/server'; + +// A settings file or plugin vault that exists but doesn't decrypt (written +// with another session secret) used to read as "nothing stored", and the +// next save wrote over it: the user's real data was gone for good. Now the +// save first moves it aside (`.unreadable-