mirror of
https://github.com/bulwarkmail/webmail.git
synced 2026-11-01 16:12:02 +00:00
fix: show plugin approval requests and decisions to every part of the server at once
This commit is contained in:
Linus Rath
@@ -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, unknown>)[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' });
|
||||
});
|
||||
});
|
||||
+155
-89
@@ -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<void> {
|
||||
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<Uint8Array | null> {
|
||||
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<void> {
|
||||
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<void> | 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<void> | null;
|
||||
}
|
||||
const GLOBAL_KEY = Symbol.for('bulwark.admin.pluginApprovals');
|
||||
|
||||
async function loadFromDisk(): Promise<ApprovalsFile> {
|
||||
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<ApprovalsFile> {
|
||||
}
|
||||
}
|
||||
|
||||
async function ensureLoaded(): Promise<void> {
|
||||
if (cached !== null) return;
|
||||
if (!loadPromise) {
|
||||
loadPromise = (async () => { cached = await loadFromDisk(); })();
|
||||
async function loadStored(): Promise<ApprovalsFile> {
|
||||
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<void> = Promise.resolve();
|
||||
|
||||
function flushToDisk(): Promise<void> {
|
||||
const run = flushQueue.then(writeToDisk, writeToDisk);
|
||||
flushQueue = run.catch(() => {});
|
||||
return run;
|
||||
async function ensureLoaded(): Promise<ApprovalsFile> {
|
||||
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<void> {
|
||||
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<boolean> {
|
||||
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<ApprovalEntry[]> {
|
||||
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<ApprovalEntry> {
|
||||
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<ApprovalEntry> {
|
||||
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<void> {
|
||||
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;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user