feat: dispatch plugin HTTP routes and wire the runtime into startup

ctx.http.route now registers a real router on the /plugin-api/:pluginId
dispatcher, replacing its stub 404. A plugin never touches the Express
req/res: it gets a plain summary and returns a plain object, so sockets
and session cookies stay on this side of the boundary. Authorization and
cookie headers are not forwarded.

Plugins load last in the start-up sequence, after the server is fully
wired, and are terminated before the database on shutdown so none can
outlive it. A plugin that fails to load cannot stop the backend.

fix: correct the nginx plugin-api location regex

The pattern was ^/plugin-api/(/.*)?$, which needs a double slash and so
never matched /plugin-api/<id>/<route>. Every other block in the file
uses ^/prefix(/.*)?$. Left alone, plugin routes would work in dev and
404 only behind Docker.
This commit is contained in:
LukeGus committed 2026-09-21 07:19:02 -05:00
1 parent 4311a0e98a
commit 4e1dec8435
7 files changed
+695 -5

No files matched your search

@@ -1,8 +1,8 @@
// Mounted at /plugin-api. Every installed plugin that exposes backend routes
// registers a sub-router here under its own plugin id, and this dispatcher
// forwards matching requests to it. Nothing registers yet, this only proves
// the path resolves end to end (nginx -> Express -> 404) ahead of the plugin
// loader landing.
// forwards matching requests to it. The plugin runtime populates the map via
// registerPluginRouter when a plugin activates, and clears it on deactivate,
// so an inactive plugin falls back to the 404 below.
//
// WebSocket note: plugin WS traffic does NOT get its own port from the
// 30001-30006 range. It rides the existing WS servers (e.g. the terminal or
@@ -18,6 +18,33 @@ const router = express.Router();
const activePluginRouters = new Map<string, Router>();
/**
* Called by the plugin runtime when a plugin activates. Replacing an existing
* entry is normal: a crash-restart re-registers under the same id.
*/
export function registerPluginRouter(
pluginId: string,
pluginRouter: Router,
): void {
activePluginRouters.set(pluginId, pluginRouter);
databaseLogger.info("Registered plugin API router", {
operation: "plugin_api_register",
pluginId,
});
}
export function unregisterPluginRouter(pluginId: string): void {
if (!activePluginRouters.delete(pluginId)) return;
databaseLogger.info("Unregistered plugin API router", {
operation: "plugin_api_register",
pluginId,
});
}
export function getRegisteredPluginIds(): string[] {
return [...activePluginRouters.keys()];
}
/**
* @openapi
* /plugin-api/{pluginId}/{path}:
+108
View File
@@ -0,0 +1,108 @@
/**
* Turns a plugin's registered routes into an Express router mounted on the
* existing /plugin-api/:pluginId dispatcher.
*
* The plugin never sees the Express req/res objects. It gets a plain,
* structured-cloneable summary of the request, and returns a plain object that
* this module turns back into a response. That keeps sockets, headers carrying
* session cookies, and the rest of the server's request context on this side of
* the boundary.
*/
import express, { type Request, type Response, type Router } from "express";
import { pluginLogger } from "../utils/logger.js";
import type { PluginBroker, PluginRuntime } from "./broker.js";
const MAX_RESPONSE_BYTES = 2 * 1024 * 1024;
/** Headers a plugin is allowed to see. Anything carrying auth is withheld. */
const FORWARDED_HEADERS = ["content-type", "accept", "user-agent"];
export function buildPluginRouter(
broker: PluginBroker,
runtime: PluginRuntime,
): Router {
const router = express.Router();
router.use(express.json({ limit: "1mb" }));
for (const route of runtime.routes) {
const key = `${route.method} ${route.path}`;
const method = route.method.toLowerCase() as
"get" | "post" | "put" | "patch" | "delete";
router[method](route.path, async (req: Request, res: Response) => {
try {
const result = await broker.invokeRoute(runtime, key, {
method: route.method,
path: req.path,
params: req.params,
query: req.query,
body: req.body ?? null,
headers: pickHeaders(req),
// The acting user, resolved by the server's own auth middleware.
// A plugin cannot set or spoof this.
userId: (req as Request & { userId?: string }).userId ?? null,
});
sendPluginResponse(res, result);
} catch (error) {
const message = error instanceof Error ? error.message : String(error);
pluginLogger.error(
`Plugin ${runtime.plugin.id} route ${key} failed`,
error instanceof Error ? error : new Error(message),
{ operation: "plugin_http" },
);
res.status(502).json({ error: "Plugin route failed", detail: message });
}
});
}
return router;
}
function pickHeaders(req: Request): Record<string, string> {
const headers: Record<string, string> = {};
for (const name of FORWARDED_HEADERS) {
const value = req.headers[name];
if (typeof value === "string") headers[name] = value;
}
return headers;
}
function sendPluginResponse(res: Response, result: unknown): void {
if (result === null || result === undefined) {
res.status(204).end();
return;
}
const shape = result as {
status?: unknown;
body?: unknown;
headers?: Record<string, unknown>;
};
// An envelope is only recognised when `status` is present. Keying off `body`
// alone would swallow an ordinary payload that happens to have a body field,
// which is a perfectly normal thing for a handler to return.
const hasEnvelope =
typeof shape === "object" && shape !== null && "status" in shape;
const status = hasEnvelope ? Number(shape.status ?? 200) : 200;
const body = hasEnvelope ? (shape.body ?? null) : result;
if (!Number.isInteger(status) || status < 100 || status > 599) {
res.status(502).json({ error: "Plugin returned an invalid status code" });
return;
}
const serialised = JSON.stringify(body);
if (
serialised &&
Buffer.byteLength(serialised, "utf8") > MAX_RESPONSE_BYTES
) {
res.status(502).json({ error: "Plugin response exceeded the size limit" });
return;
}
res.status(status).json(body);
}
+211
View File
@@ -0,0 +1,211 @@
/**
* Composition root for the plugin runtime: wires the loader, the broker and the
* /plugin-api dispatcher together and exposes the handful of calls the rest of
* the server needs.
*
* The heavy dependencies (host resolution, the SSH pool, the repositories) are
* imported lazily inside the dep functions rather than at module scope. Several
* of them pull in the repository layer, and a static import here would create
* the same cycle hosts/automation-events.ts documents.
*/
import { pluginLogger } from "../utils/logger.js";
import {
registerPluginRouter,
unregisterPluginRouter,
} from "../database/routes/plugin-api-routes.js";
import { PluginBroker, type PluginRuntime } from "./broker.js";
import { buildPluginRouter } from "./http-bridge.js";
import { PluginLoader, type LoadedPlugin } from "./loader.js";
let loader: PluginLoader | null = null;
let broker: PluginBroker | null = null;
function createBroker(): PluginBroker {
return new PluginBroker({
listHosts: async (userId) => {
const { createCurrentHostRepository } =
await import("../database/repositories/factory.js");
// listByUserId, not listDecryptedByUserId: a plugin only ever gets host
// metadata, so there is no reason to decrypt secrets just to drop them.
const hosts = await createCurrentHostRepository().listByUserId(userId);
return hosts as unknown as Record<string, unknown>[];
},
resolveHost: async (hostId, userId) => {
// resolveHostById is the access gate and the decryption step in one. The
// object it returns carries plaintext secrets, so it must never leave
// this closure -- the broker maps it through toPluginHostView first.
const { resolveHostById } = await import("../hosts/host-resolver.js");
const host = await resolveHostById(hostId, userId);
return host as unknown as Record<string, unknown> | null;
},
storage: {
get: async (pluginId, key) => {
const { createCurrentPluginStorageRepository } =
await import("../database/repositories/factory.js");
return createCurrentPluginStorageRepository().get(pluginId, key);
},
set: async (pluginId, key, value) => {
const { createCurrentPluginStorageRepository } =
await import("../database/repositories/factory.js");
await createCurrentPluginStorageRepository().set(pluginId, key, value);
},
delete: async (pluginId, key) => {
const { createCurrentPluginStorageRepository } =
await import("../database/repositories/factory.js");
return createCurrentPluginStorageRepository().delete(pluginId, key);
},
listKeys: async (pluginId) => {
const { createCurrentPluginStorageRepository } =
await import("../database/repositories/factory.js");
return createCurrentPluginStorageRepository().listKeys(pluginId);
},
},
ssh: {
/**
* The single most security-sensitive function in the plugin runtime.
*
* `host` below is the object resolveHostById returns: it carries the
* plaintext password, private key and key passphrase. It is a local
* inside this closure and is referenced only by createFleetSshFactory.
* Nothing derived from it is returned. The caller gets two closures --
* release and exec -- and exec resolves to stdout/stderr/code strings
* only, so there is no path from the worker back to the credential.
*/
open: async (hostId, userId) => {
const [
{ resolveHostById },
{ connectionPool },
{ createFleetSshFactory, getFleetPoolKey },
{ execCommand },
] = await Promise.all([
import("../hosts/host-resolver.js"),
import("../hosts/ssh-connection-pool.js"),
import("../hosts/ssh-client-factory.js"),
import("../hosts/metrics/widgets/common-utils.js"),
]);
// resolveHostById performs the RBAC check (canAccessHost) and returns
// null for both "missing" and "forbidden".
const host = await resolveHostById(hostId, userId);
if (!host) {
throw new Error(`Host ${hostId} not found or not accessible`);
}
const poolKey = getFleetPoolKey(host);
const client = await connectionPool.getConnection(
poolKey,
createFleetSshFactory(host),
);
return {
release: () => connectionPool.releaseConnection(poolKey, client),
exec: (command, timeoutMs) => execCommand(client, command, timeoutMs),
};
},
},
onRoutesChanged: (runtime) => remountRoutes(runtime),
});
}
function remountRoutes(runtime: PluginRuntime): void {
if (runtime.routes.length === 0) return;
registerPluginRouter(runtime.plugin.id, buildPluginRouter(broker!, runtime));
}
export function getPluginRuntime(): {
loader: PluginLoader;
broker: PluginBroker;
} {
if (!loader || !broker) {
broker = createBroker();
loader = new PluginLoader({
onWorkerReady: (plugin, worker) => broker!.attach(plugin, worker),
onWorkerGone: (plugin) => {
broker!.detach(plugin.id);
unregisterPluginRouter(plugin.id);
},
});
}
return { loader, broker };
}
/**
* Loads every plugin on disk and activates the ones marked enabled in the
* database. Called once from the backend start-up sequence.
*/
export async function initializePlugins(): Promise<LoadedPlugin[]> {
const { loader: pluginLoader } = getPluginRuntime();
const loaded = await pluginLoader.loadAll();
if (loaded.length === 0) return [];
const {
createCurrentPluginRepository,
createCurrentPluginPermissionGrantRepository,
} = await import("../database/repositories/factory.js");
const records = await createCurrentPluginRepository().listAll();
const byId = new Map(records.map((record) => [record.id, record]));
for (const plugin of loaded) {
const record = byId.get(plugin.id);
if (!record || record.state !== "enabled") continue;
// The plugins table records no installer, so the plugin acts under the
// authority of whoever granted its capabilities. Without a grant there is
// no owner and nothing the plugin could do with one, so it stays stopped.
const grants =
await createCurrentPluginPermissionGrantRepository().listByPlugin(
plugin.id,
);
const ownerUserId = grants[0]?.grantedBy;
if (!ownerUserId) {
pluginLogger.warn(
`Plugin ${plugin.id} is enabled but has no capability grants, leaving it stopped`,
{ operation: "plugin_activate" },
);
continue;
}
try {
await activatePlugin(plugin.id, ownerUserId);
} catch (error) {
pluginLogger.error(
`Failed to activate plugin ${plugin.id}`,
error instanceof Error ? error : new Error(String(error)),
{ operation: "plugin_activate" },
);
}
}
return loaded;
}
export async function activatePlugin(
pluginId: string,
ownerUserId: string,
): Promise<void> {
const { loader: pluginLoader, broker: pluginBroker } = getPluginRuntime();
await pluginLoader.activate(pluginId, ownerUserId);
const runtime = pluginBroker.get(pluginId);
if (runtime) remountRoutes(runtime);
}
export async function deactivatePlugin(pluginId: string): Promise<void> {
const { loader: pluginLoader } = getPluginRuntime();
await pluginLoader.deactivate(pluginId);
unregisterPluginRouter(pluginId);
}
export async function shutdownPlugins(): Promise<void> {
if (!loader) return;
for (const plugin of loader.list()) unregisterPluginRouter(plugin.id);
await loader.shutdown();
}
+27
View File
@@ -343,6 +343,23 @@ async function provisionLocalDesktopUserIfNeeded(): Promise<void> {
await import("./automations/scheduler.js");
startAutomationScheduler();
// Last, so a plugin's activate() sees a fully wired server. A plugin that
// fails to load must not stop the backend, so this never rejects.
try {
const { initializePlugins } = await import("./plugins/index.js");
const loaded = await initializePlugins();
if (loaded.length > 0) {
systemLogger.info(`Loaded ${loaded.length} plugin(s)`, {
operation: "plugin_init",
});
}
} catch (error) {
systemLogger.warn("Plugin runtime failed to initialize", {
operation: "plugin_init",
error: getErrorMessage(error),
});
}
const { startAnalyticsHeartbeat } = await import("./utils/analytics.js");
startAnalyticsHeartbeat();
@@ -357,6 +374,16 @@ async function provisionLocalDesktopUserIfNeeded(): Promise<void> {
systemLogger.info(`Received ${signal}, initiating graceful shutdown...`, {
operation: "shutdown",
});
// Terminate plugin workers before the database goes away, so a plugin
// mid-write cannot outlive it.
try {
const { shutdownPlugins } = await import("./plugins/index.js");
await shutdownPlugins();
} catch {
// Nothing to stop.
}
// Only SQLite has anything to flush. On a client-server engine the writes
// committed as they happened, so there is no file to save and claiming
// otherwise in the log would be untrue.
@@ -0,0 +1,317 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import express from "express";
import type { AddressInfo } from "node:net";
import type { Server } from "node:http";
import { createFixturePlugin, type Fixture } from "./fixture-plugin.js";
const state = vi.hoisted(() => ({
grants: [] as { pluginId: string; capability: string }[],
}));
vi.mock("../../utils/logger.js", () => ({
pluginLogger: {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
success: vi.fn(),
},
databaseLogger: {
debug: vi.fn(),
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
success: vi.fn(),
},
}));
vi.mock("../../utils/audit-logger.js", () => ({
logAudit: vi.fn(async () => {}),
getAuditUsername: vi.fn(async (userId: string) => `user:${userId}`),
}));
vi.mock("../../database/repositories/factory.js", () => ({
createCurrentPluginPermissionGrantRepository: () => ({
listByPlugin: async (pluginId: string) =>
state.grants.filter((g) => g.pluginId === pluginId),
}),
}));
const { PluginBroker } = await import("../../plugins/broker.js");
const { PluginLoader } = await import("../../plugins/loader.js");
const { buildPluginRouter } = await import("../../plugins/http-bridge.js");
const pluginApi = await import("../../database/routes/plugin-api-routes.js");
const WORKER_TIMEOUT = 30_000;
describe("plugin HTTP bridge", () => {
const fixtures: Fixture[] = [];
let loader: InstanceType<typeof PluginLoader> | null = null;
let broker: InstanceType<typeof PluginBroker> | null = null;
let server: Server | null = null;
let baseUrl = "";
beforeEach(async () => {
state.grants = [];
const app = express();
app.use("/plugin-api", pluginApi.default);
server = await new Promise<Server>((resolve) => {
const created = app.listen(0, "127.0.0.1", () => resolve(created));
});
baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}`;
});
afterEach(async () => {
await loader?.shutdown();
loader = null;
broker = null;
for (const id of pluginApi.getRegisteredPluginIds()) {
pluginApi.unregisterPluginRouter(id);
}
await new Promise<void>((resolve) => server?.close(() => resolve()));
server = null;
while (fixtures.length) fixtures.pop()!.cleanup();
});
async function activate(backendSource: string) {
const fixture = createFixturePlugin({ backendSource });
fixtures.push(fixture);
broker = new PluginBroker({
listHosts: async () => [],
resolveHost: async () => null,
onRoutesChanged: (runtime) => {
pluginApi.registerPluginRouter(
runtime.plugin.id,
buildPluginRouter(broker!, runtime),
);
},
});
loader = new PluginLoader({
onWorkerReady: (plugin, worker) => broker!.attach(plugin, worker),
onWorkerGone: (plugin) => {
broker!.detach(plugin.id);
pluginApi.unregisterPluginRouter(plugin.id);
},
});
await loader.load(fixture.dir);
await loader.activate("sample-plugin", "user-1");
}
it("404s for a plugin that is not running", async () => {
const res = await fetch(`${baseUrl}/plugin-api/ghost/anything`);
expect(res.status).toBe(404);
expect(await res.json()).toEqual({
error: "Plugin not installed or not running",
});
});
it(
"dispatches a GET to an active plugin's handler",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/ping", async () => ({ pong: true }));
}
`);
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/ping`);
expect(res.status).toBe(200);
expect(await res.json()).toEqual({ pong: true });
},
WORKER_TIMEOUT,
);
it(
"passes the body, query and params through to the handler",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("POST", "/echo/:id", async (req) => ({
id: req.params.id,
q: req.query.q,
body: req.body,
method: req.method,
}));
}
`);
const res = await fetch(
`${baseUrl}/plugin-api/sample-plugin/echo/7?q=hello`,
{
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({ name: "termix" }),
},
);
expect(await res.json()).toEqual({
id: "7",
q: "hello",
body: { name: "termix" },
method: "POST",
});
},
WORKER_TIMEOUT,
);
it(
"honours an explicit status/body envelope",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/missing", async () => ({
status: 404,
body: { error: "nope" },
}));
}
`);
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/missing`);
expect(res.status).toBe(404);
expect(await res.json()).toEqual({ error: "nope" });
},
WORKER_TIMEOUT,
);
it(
"treats a payload with a body field as the body, not an envelope",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/doc", async () => ({
body: "the article text",
title: "A post",
}));
}
`);
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/doc`);
expect(res.status).toBe(200);
expect(await res.json()).toEqual({
body: "the article text",
title: "A post",
});
},
WORKER_TIMEOUT,
);
it(
"returns 502 when the plugin handler throws",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/boom", async () => {
throw new Error("handler exploded");
});
}
`);
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/boom`);
expect(res.status).toBe(502);
expect((await res.json()).detail).toContain("handler exploded");
},
WORKER_TIMEOUT,
);
it(
"does not forward authorization headers or cookies to the plugin",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/headers", async (req) => ({
seen: Object.keys(req.headers).sort(),
}));
}
`);
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/headers`, {
headers: {
authorization: "Bearer super-secret-token",
cookie: "session=secret",
"content-type": "application/json",
},
});
const seen = (await res.json()).seen as string[];
expect(seen).not.toContain("authorization");
expect(seen).not.toContain("cookie");
expect(seen).toContain("content-type");
},
WORKER_TIMEOUT,
);
it(
"rejects an unsupported method and a path without a leading slash",
async () => {
const fixture = createFixturePlugin({
backendSource: `
export async function activate(ctx) {
const errors = [];
try {
await ctx.http.route("TRACE", "/x", () => null);
} catch (e) { errors.push(e.message); }
try {
await ctx.http.route("GET", "no-slash", () => null);
} catch (e) { errors.push(e.message); }
if (errors.length !== 2) throw new Error("expected two rejections");
await ctx.http.route("GET", "/ok", async () => ({ errors }));
}
`,
});
fixtures.push(fixture);
broker = new PluginBroker({
listHosts: async () => [],
resolveHost: async () => null,
onRoutesChanged: (runtime) => {
pluginApi.registerPluginRouter(
runtime.plugin.id,
buildPluginRouter(broker!, runtime),
);
},
});
loader = new PluginLoader({
onWorkerReady: (plugin, worker) => broker!.attach(plugin, worker),
onWorkerGone: (plugin) => broker!.detach(plugin.id),
});
await loader.load(fixture.dir);
await loader.activate("sample-plugin", "user-1");
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/ok`);
const errors = (await res.json()).errors as string[];
expect(errors[0]).toContain("does not support the TRACE method");
expect(errors[1]).toContain('must start with "/"');
},
WORKER_TIMEOUT,
);
it(
"404s again once the plugin is deactivated",
async () => {
await activate(`
export async function activate(ctx) {
await ctx.http.route("GET", "/ping", async () => ({ pong: true }));
}
`);
expect(
(await fetch(`${baseUrl}/plugin-api/sample-plugin/ping`)).status,
).toBe(200);
await loader!.deactivate("sample-plugin");
pluginApi.unregisterPluginRouter("sample-plugin");
const res = await fetch(`${baseUrl}/plugin-api/sample-plugin/ping`);
expect(res.status).toBe(404);
},
WORKER_TIMEOUT,
);
});