From e75834314df2f45d753384935cedc7e43917b765 Mon Sep 17 00:00:00 2001 From: Linus Rath <139418639+rathlinus@users.noreply.github.com> Date: Sun, 1 Nov 2026 00:13:08 +0100 Subject: [PATCH] fix: show plugin approval requests and decisions to every part of the server at once --- .../__tests__/plugin-approvals-copies.test.ts | 48 ++++ lib/admin/plugin-approvals.ts | 244 +++++++++++------- 2 files changed, 203 insertions(+), 89 deletions(-) create mode 100644 lib/admin/__tests__/plugin-approvals-copies.test.ts diff --git a/lib/admin/__tests__/plugin-approvals-copies.test.ts b/lib/admin/__tests__/plugin-approvals-copies.test.ts new file mode 100644 index 000000000..e3c97c03d --- /dev/null +++ b/lib/admin/__tests__/plugin-approvals-copies.test.ts @@ -0,0 +1,48 @@ +// @vitest-environment node +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; + +// Next makes one copy of plugin-approvals.ts per bundle entry. Each copy +// loaded the approvals file once and kept it: a request taken by the +// approval-status route never reached the admin route's copy, a decision +// there never reached the plugin-secrets gate, and the next write of any +// copy put its old list back over the others' changes. + +const HASH = 'b'.repeat(64); +let dir: string; + +beforeEach(async () => { + dir = await mkdtemp(path.join(tmpdir(), 'bw-approvals-copies-')); + vi.stubEnv('ADMIN_CONFIG_DIR', dir); + delete (globalThis as Record)[Symbol.for('bulwark.admin.pluginApprovals')]; +}); + +afterEach(async () => { + vi.unstubAllEnvs(); + await rm(dir, { recursive: true, force: true }); +}); + +async function moduleCopy() { + vi.resetModules(); + return import('../plugin-approvals'); +} + +describe('plugin approvals across module copies', () => { + it('shows a request and its decision to every copy at once', async () => { + const statusRoute = await moduleCopy(); + const adminRoute = await moduleCopy(); + const gate = await moduleCopy(); + // Each copy has read the (empty) file already. + expect(await adminRoute.listApprovals()).toEqual([]); + expect(await gate.listApprovals()).toEqual([]); + + await statusRoute.requestApproval('cloud-sync', HASH, { name: 'Cloud' }, 'alice@example.org'); + expect((await adminRoute.listApprovals()).map((e) => e.status)).toEqual(['pending']); + + await adminRoute.decideApproval('cloud-sync', HASH, 'approved', 'admin'); + expect(await gate.getApprovalStatus('cloud-sync', HASH)).toMatchObject({ status: 'approved' }); + expect(await statusRoute.getApprovalStatus('cloud-sync', HASH)).toMatchObject({ status: 'approved' }); + }); +}); diff --git a/lib/admin/plugin-approvals.ts b/lib/admin/plugin-approvals.ts index 532fe7cdd..462bc114b 100644 --- a/lib/admin/plugin-approvals.ts +++ b/lib/admin/plugin-approvals.ts @@ -9,12 +9,19 @@ // Each entry has one of three states: 'pending' (user installed, waiting for // admin), 'approved' (admin signed off), 'denied' (admin refused - kept so we // don't keep asking). +// +// Stored in C/plugin-approvals.json, or the shared store in shared mode +// (#1066). Every change goes through update() against the stored document, +// so requests and decisions arriving at once (on one instance or several) +// all stay, then announces area `approvals`: every replica re-reads its copy, +// which also feeds the plugin-secrets credential gate (known-plugins.ts). -import { readFile, writeFile, rename, mkdir, unlink } from 'node:fs/promises'; -import nodePath from 'node:path'; -import { existsSync } from 'node:fs'; import { logger } from '@/lib/logger'; -import { ensureConfigDir, getConfigPath, assertWritable } from './paths'; +import { isSharedMode } from '@/lib/shared-store'; +import { announceChange, onSharedChange } from '@/lib/shared-store/changes'; +import { ensureConfigDir, assertWritable } from './paths'; +import { isStoreUnavailable } from './admin-store'; +import { pluginApprovalDocs } from './extension-store'; import type { CredentialSpec } from '@/lib/plugin-secrets/credentials'; export type ApprovalStatus = 'pending' | 'approved' | 'denied'; @@ -51,38 +58,54 @@ const MAX_ENTRIES = 500; // hard cap so a misbehaving client can't grow the file export const MAX_PENDING_PER_USER = 10; /** The package a user asked to have approved, kept so the admin can read the code. */ -function bundlePath(pluginId: string, bundleHash: string): string { - return nodePath.join(getConfigPath('plugin-approvals'), `${pluginId}-${bundleHash}.zip`); +function bundleId(pluginId: string, bundleHash: string): string { + return `plugin-approvals/${pluginId}-${bundleHash}.zip`; } export async function saveApprovalBundle(pluginId: string, bundleHash: string, zip: Uint8Array): Promise { assertWritable('plugin-approvals.saveBundle'); - await mkdir(getConfigPath('plugin-approvals'), { recursive: true }); - await writeFile(bundlePath(pluginId, bundleHash), zip); + await pluginApprovalDocs().put(bundleId(pluginId, bundleHash), Buffer.from(zip)); } export async function readApprovalBundle(pluginId: string, bundleHash: string): Promise { try { - return new Uint8Array(await readFile(bundlePath(pluginId, bundleHash))); - } catch { + const doc = await pluginApprovalDocs().get(bundleId(pluginId, bundleHash)); + return doc ? new Uint8Array(doc.data) : null; + } catch (error) { + if (isStoreUnavailable(error)) throw error; return null; } } async function deleteApprovalBundle(pluginId: string, bundleHash: string): Promise { - try { await unlink(bundlePath(pluginId, bundleHash)); } catch { /* none */ } + try { await pluginApprovalDocs().delete(bundleId(pluginId, bundleHash)); } catch { /* none */ } } -let cached: ApprovalsFile | null = null; -let loadPromise: Promise | null = null; +// The copy every module copy (admin routes, the approval-status route, the +// plugin-secrets gate) reads, on globalThis. In shared mode another +// replica's change marks it stale; while the store is down the stale copy +// keeps being served. +interface ApprovalsGlobals { + cached: ApprovalsFile | null; + stale: boolean; + loading: Promise | null; +} +const GLOBAL_KEY = Symbol.for('bulwark.admin.pluginApprovals'); -async function loadFromDisk(): Promise { - await ensureConfigDir(); - const path = getConfigPath(APPROVALS_FILE); - if (!existsSync(path)) return { entries: [] }; +function approvalsGlobals(): ApprovalsGlobals { + const g = globalThis as typeof globalThis & { [GLOBAL_KEY]?: ApprovalsGlobals }; + return (g[GLOBAL_KEY] ??= { cached: null, stale: false, loading: null }); +} + +onSharedChange('approvals', (origin) => { + // This process's own writes already replaced the copy. + if (origin === 'remote') approvalsGlobals().stale = true; +}); + +function parseFile(data: Buffer | null): ApprovalsFile { + if (!data) return { entries: [] }; try { - const raw = await readFile(path, 'utf-8'); - const parsed = JSON.parse(raw); + const parsed = JSON.parse(data.toString('utf-8')); if (!parsed || !Array.isArray(parsed.entries)) return { entries: [] }; return { entries: parsed.entries.filter(isWellFormed) }; } catch (err) { @@ -91,33 +114,57 @@ async function loadFromDisk(): Promise { } } -async function ensureLoaded(): Promise { - if (cached !== null) return; - if (!loadPromise) { - loadPromise = (async () => { cached = await loadFromDisk(); })(); +async function loadStored(): Promise { + if (!isSharedMode()) await ensureConfigDir(); + let data: Buffer | null; + try { + data = (await pluginApprovalDocs().get(APPROVALS_FILE))?.data ?? null; + } catch (err) { + if (isStoreUnavailable(err)) throw err; + logger.warn('[plugin-approvals] failed to read file', { error: err instanceof Error ? err.message : String(err) }); + return { entries: [] }; } - await loadPromise; + return parseFile(data); } -// Writes run one after another. Requests of several users at once shared the -// one ".tmp" file, and every rename after the first failed: the user got a -// 500 and the server kept no ZIP for the admin to review. -let flushQueue: Promise = Promise.resolve(); - -function flushToDisk(): Promise { - const run = flushQueue.then(writeToDisk, writeToDisk); - flushQueue = run.catch(() => {}); - return run; +async function ensureLoaded(): Promise { + const g = approvalsGlobals(); + if (g.cached && !g.stale) return g.cached; + if (!g.loading) { + g.stale = false; + g.loading = (async () => { + try { + g.cached = await loadStored(); + } catch (err) { + // Store down: keep serving the copy we have, read again next time. + if (!(isStoreUnavailable(err) && g.cached)) throw err; + g.stale = true; + } finally { + g.loading = null; + } + })(); + } + await g.loading; + return g.cached!; } -async function writeToDisk(): Promise { - if (!cached) return; - await ensureConfigDir(); - assertWritable('plugin-approvals.flushToDisk'); - const path = getConfigPath(APPROVALS_FILE); - const tmp = `${path}.tmp`; - await writeFile(tmp, JSON.stringify(cached, null, 2), 'utf-8'); - await rename(tmp, path); +/** + * Change the stored file with `change`, which edits the parsed file and + * returns true when it changed something. It may run more than once. + */ +async function changeStored(change: (file: ApprovalsFile) => boolean): Promise { + let changed = false; + const result = await pluginApprovalDocs().update(APPROVALS_FILE, (current) => { + const file = parseFile(current); + changed = change(file); + return changed ? JSON.stringify(file, null, 2) : undefined; + }); + if (changed && result.data) { + approvalsGlobals().cached = parseFile(result.data); + approvalsGlobals().stale = false; + await announceChange('approvals'); + } + return changed; } function isWellFormed(value: unknown): value is ApprovalEntry { @@ -140,13 +187,11 @@ function findEntry(file: ApprovalsFile, pluginId: string, bundleHash: string): A // ─── Public API ────────────────────────────────────────────── export async function listApprovals(): Promise { - await ensureLoaded(); - return [...cached!.entries]; + return [...(await ensureLoaded()).entries]; } export async function getApprovalStatus(pluginId: string, bundleHash: string): Promise<{ status: ApprovalStatus | 'not-requested'; decidedAt?: string }> { - await ensureLoaded(); - const entry = findEntry(cached!, pluginId, bundleHash); + const entry = findEntry(await ensureLoaded(), pluginId, bundleHash); if (!entry) return { status: 'not-requested' }; return { status: entry.status, decidedAt: entry.decidedAt }; } @@ -158,37 +203,47 @@ export async function requestApproval( requestedBy: string, ): Promise { if (!pluginId || !bundleHash) throw new Error('pluginId and bundleHash required'); - await ensureLoaded(); - const file = cached!; - const existing = findEntry(file, pluginId, bundleHash); + const existing = findEntry(await ensureLoaded(), pluginId, bundleHash); if (existing) return existing; assertWritable('request plugin approval'); - const pendingByUser = file.entries.filter(e => e.status === 'pending' && e.requestedBy === requestedBy).length; - if (pendingByUser >= MAX_PENDING_PER_USER) { - throw new ApprovalLimitError(`You already have ${MAX_PENDING_PER_USER} plugins waiting for approval`); - } - if (file.entries.length >= MAX_ENTRIES) { - // Drop the oldest pending entry so a new request can land. Approved/denied - // entries are preserved. - const oldestPendingIdx = file.entries.findIndex(e => e.status === 'pending'); - if (oldestPendingIdx >= 0) { - const [dropped] = file.entries.splice(oldestPendingIdx, 1); - await deleteApprovalBundle(dropped.pluginId, dropped.bundleHash); - } else { - throw new ApprovalLimitError('The approval queue is full; ask an administrator to clear old requests'); + const requestedAt = new Date().toISOString(); + let result: ApprovalEntry | null = null; + let dropped: ApprovalEntry | null = null; + await changeStored((file) => { + dropped = null; + const stored = findEntry(file, pluginId, bundleHash); + if (stored) { + result = stored; + return false; } - } - const entry: ApprovalEntry = { - pluginId, - bundleHash, - status: 'pending', - manifest, - requestedBy, - requestedAt: new Date().toISOString(), - }; - file.entries.push(entry); - await flushToDisk(); - return entry; + const pendingByUser = file.entries.filter(e => e.status === 'pending' && e.requestedBy === requestedBy).length; + if (pendingByUser >= MAX_PENDING_PER_USER) { + throw new ApprovalLimitError(`You already have ${MAX_PENDING_PER_USER} plugins waiting for approval`); + } + if (file.entries.length >= MAX_ENTRIES) { + // Drop the oldest pending entry so a new request can land. Approved/denied + // entries are preserved. + const oldestPendingIdx = file.entries.findIndex(e => e.status === 'pending'); + if (oldestPendingIdx < 0) { + throw new ApprovalLimitError('The approval queue is full; ask an administrator to clear old requests'); + } + [dropped] = file.entries.splice(oldestPendingIdx, 1); + } + const entry: ApprovalEntry = { + pluginId, + bundleHash, + status: 'pending', + manifest, + requestedBy, + requestedAt, + }; + file.entries.push(entry); + result = entry; + return true; + }); + const gone = dropped as ApprovalEntry | null; + if (gone) await deleteApprovalBundle(gone.pluginId, gone.bundleHash); + return result!; } export async function decideApproval( @@ -198,26 +253,37 @@ export async function decideApproval( decidedBy: string, ): Promise { assertWritable('decide plugin approval'); - await ensureLoaded(); - const file = cached!; - const entry = findEntry(file, pluginId, bundleHash); - if (!entry) throw new Error('approval entry not found'); - entry.status = decision; - entry.decidedAt = new Date().toISOString(); - entry.decidedBy = decidedBy; - await flushToDisk(); - return entry; + const decidedAt = new Date().toISOString(); + let decided: ApprovalEntry | null = null; + await changeStored((file) => { + const entry = findEntry(file, pluginId, bundleHash); + if (!entry) throw new Error('approval entry not found'); + entry.status = decision; + entry.decidedAt = decidedAt; + entry.decidedBy = decidedBy; + decided = entry; + return true; + }); + return decided!; } export async function revokeApproval(pluginId: string, bundleHash: string): Promise { assertWritable('revoke plugin approval'); - await ensureLoaded(); - const file = cached!; - const idx = file.entries.findIndex(e => e.pluginId === pluginId && e.bundleHash === bundleHash); - if (idx < 0) return; - file.entries.splice(idx, 1); - await flushToDisk(); - await deleteApprovalBundle(pluginId, bundleHash); + const removed = await changeStored((file) => { + const idx = file.entries.findIndex(e => e.pluginId === pluginId && e.bundleHash === bundleHash); + if (idx < 0) return false; + file.entries.splice(idx, 1); + return true; + }); + if (removed) await deleteApprovalBundle(pluginId, bundleHash); } export class ApprovalLimitError extends Error {} + +/** Forget the copy in memory (tests: a fresh process). */ +export function resetApprovalsCacheForTests(): void { + const g = approvalsGlobals(); + g.cached = null; + g.stale = false; + g.loading = null; +}