feat: serve plugin http and websockets from one mount instead of ports

This commit is contained in:
LukeGus committed 2026-09-22 16:39:41 -05:00
1 parent c496a093ca
commit 8431570ff7
102 files changed
+2894 -2308

No files matched your search

+86 -1
View File
@@ -29,8 +29,18 @@ import type { SecretRegistration } from "./secret-registry.js";
import { assertCapability } from "./permissions.js";
import { getActor, runAsActor } from "./actor.js";
import { DisposableBag } from "./disposables.js";
import {
createPluginRouter,
createRbacMiddleware,
unregisterPluginHttp,
} from "./http.js";
import { registerPluginWsRoute, registerPluginWsUpgrade } from "./ws.js";
import type { PluginManifest } from "@termix/plugin-sdk/manifest";
import type { PluginContext, PluginModule } from "@termix/plugin-sdk/backend";
import {
PluginCapabilityError,
type PluginContext,
type PluginModule,
} from "@termix/plugin-sdk/backend";
import type { PluginTableDefinition } from "@termix/plugin-sdk/db";
import * as syncRegistry from "./sync-registry.js";
@@ -410,6 +420,63 @@ export function createPluginContext(
}),
},
http: {
router: (options) => {
// Declared here, granted per request: the grant lives in the database
// and router() has to be synchronous so a plugin can register routes
// inline in activate. http.ts re-checks on every request.
if (!declared.includes("network:serve")) {
throw new PluginCapabilityError(pluginId, "network:serve");
}
const router = createPluginRouter({
manifest,
options,
reportError: (error) => void reportRuntimeError(pluginId, error),
});
handle.bag.add(
() => unregisterPluginHttp(pluginId),
`HTTP router for ${pluginId}`,
);
return router as never;
},
},
ws: {
route: (path, wsHandler, options) => {
if (!declared.includes("network:serve")) {
throw new PluginCapabilityError(pluginId, "network:serve");
}
const dispose = registerPluginWsRoute(
pluginId,
path,
wsHandler,
declared,
options,
);
handle.bag.add(dispose, `WebSocket route "${path}"`);
},
upgrade: (path, upgradeHandler, options) => {
if (!declared.includes("network:serve")) {
throw new PluginCapabilityError(pluginId, "network:serve");
}
const dispose = registerPluginWsUpgrade(
pluginId,
path,
upgradeHandler as never,
declared,
options,
);
handle.bag.add(dispose, `WebSocket upgrade "${path}"`);
},
},
rbac: {
require: (permission) => createRbacMiddleware(manifest, permission),
},
disposables: {
add: (dispose) => handle.bag.add(dispose, "plugin resource"),
},
@@ -439,6 +506,24 @@ export function createPluginContext(
};
}
/**
* Feeds a route or socket error into the loader's error budget.
*
* Lazily imported: index.ts owns the loader and imports this module, so a
* static import here would close the cycle.
*/
async function reportRuntimeError(
pluginId: string,
error: unknown,
): Promise<void> {
try {
const { getPluginRuntime } = await import("./index.js");
await getPluginRuntime().loader.reportError(pluginId, error);
} catch {
// The budget is a safety net, not a dependency of serving a request.
}
}
function assertKvKey(key: string): void {
if (typeof key !== "string" || key.length === 0) {
throw new Error("ctx.kv keys must be non-empty strings");
+6
View File
@@ -34,6 +34,12 @@ export const TOPICS = {
hostStatus: "host.status",
/** A health check result. */
hostHealthCheck: "host.health_check",
/** A host's connection details changed and pollers should re-read them. */
hostUpdated: "host.updated",
/** A host was deleted and any poller holding it should drop it. */
hostDeleted: "host.deleted",
/** Someone logged in over SSH to a host. */
hostLogin: "host.login",
} as const;
export type EventListener = (payload: unknown) => void;
+315
View File
@@ -0,0 +1,315 @@
/**
* HTTP for plugins: one mount point, one set of middleware, no ports.
*
* A plugin calls ctx.http.router() and gets an Express Router that core mounts
* at /plugin-api/<id>/. Everything a route needs in front of it runs here
* rather than in the plugin:
*
* 1. auth, unless the path was declared public;
* 2. the actor, so ctx and core APIs know who is calling;
* 3. an enabled check, so a disabled plugin answers 503 rather than 404;
* 4. body limits;
* 5. an error wrapper that logs against the plugin and leaks nothing.
*
* The wrapper never touches the response body, so SSE, file downloads and
* multipart uploads stream through untouched.
*/
import express, {
type NextFunction,
type Request,
type Response,
type Router,
} from "express";
import type {
PluginMiddleware,
PluginRouterOptions,
} from "@termix/plugin-sdk/backend";
import type { PluginManifest } from "@termix/plugin-sdk/manifest";
import { pluginLogger } from "../utils/logger.js";
import { AuthManager } from "../utils/auth-manager.js";
import { runAsActor } from "./actor.js";
import { hasCapability } from "./permissions.js";
/** Matches what core's own routes accept, so moving a route changes nothing. */
const DEFAULT_BODY_LIMIT = "2mb";
/** Routers by plugin id, read by the dispatcher in plugin-api-routes.ts. */
const routers = new Map<string, Router>();
export function getPluginRouter(pluginId: string): Router | undefined {
return routers.get(pluginId);
}
export function getRegisteredHttpPluginIds(): string[] {
return [...routers.keys()];
}
export function unregisterPluginHttp(pluginId: string): void {
if (!routers.delete(pluginId)) return;
pluginLogger.info(`Unmounted /plugin-api/${pluginId}`, {
operation: "plugin_http_unmount",
});
}
/**
* Whether the plugin is currently running.
*
* Injected rather than imported so this module does not depend on the loader,
* which imports ctx.ts, which would close a cycle.
*/
type IsEnabled = (pluginId: string) => boolean;
let isPluginEnabled: IsEnabled = () => true;
export function setPluginEnabledCheck(check: IsEnabled): void {
isPluginEnabled = check;
}
/** Test seam, so one test's stub does not leak into the next. */
export function resetPluginHttp(): void {
routers.clear();
isPluginEnabled = () => true;
}
function normalizePublicPath(path: string): string {
const trimmed = path.trim();
const withSlash = trimmed.startsWith("/") ? trimmed : `/${trimmed}`;
return withSlash.length > 1 && withSlash.endsWith("/")
? withSlash.slice(0, -1)
: withSlash;
}
/**
* Compiles a declared public path into a matcher.
*
* Express-style ":param" segments are supported, because the routes that have
* to be public are usually keyed by a token ("/webhook/:token"). A parameter
* matches exactly one segment and never a slash, so ":token" cannot swallow
* the rest of the path and turn one declared route into a prefix hole.
*/
function compilePublicPath(path: string): RegExp {
const escaped = path
.split("/")
.map((segment) => {
if (segment.startsWith(":")) return "[^/]+";
return segment.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
})
.join("/");
return new RegExp(`^${escaped}$`);
}
/**
* Matches the request path against the declared public paths.
*
* Full-path match only. A prefix match would turn one declared route into a
* hole covering everything below it, which is not what declaring a single
* public path should mean.
*/
function isPublicRequest(req: Request, publicPaths: RegExp[]): boolean {
if (publicPaths.length === 0) return false;
const path = normalizePublicPath(req.path || "/");
return publicPaths.some((pattern) => pattern.test(path));
}
export interface CreatePluginRouterArgs {
manifest: PluginManifest;
options?: PluginRouterOptions;
/** Records a runtime error against the plugin's error budget. */
reportError: (error: unknown) => void;
}
export function createPluginRouter({
manifest,
options,
reportError,
}: CreatePluginRouterArgs): Router {
const pluginId = manifest.id;
const declaredPublic = (options?.public ?? []).map((path) =>
normalizePublicPath(path),
);
const publicPaths = declaredPublic.map((path) => compilePublicPath(path));
if (declaredPublic.length > 0) {
// Audited once at registration: the set of unauthenticated routes a plugin
// has is worth seeing even if nobody ever calls them.
pluginLogger.warn(
`Plugin ${pluginId} serves ${declaredPublic.length} route(s) without authentication: ${declaredPublic.join(", ")}`,
{ operation: "plugin_http_public" },
);
void writePublicRouteAudit(manifest, declaredPublic);
}
const outer = express.Router();
const inner = express.Router();
outer.use((req: Request, res: Response, next: NextFunction) => {
if (isPluginEnabled(pluginId)) {
next();
return;
}
res.status(503).json({ error: "Plugin is not running", pluginId });
});
// The grant lives in the database, so it cannot be checked in the
// synchronous router() call a plugin makes inside activate. Checked per
// request instead: a plugin whose network:serve was revoked keeps its mount
// point and refuses every request through it, which is the same outcome a
// revoke should have and takes effect without a restart.
outer.use((req: Request, res: Response, next: NextFunction) => {
void (async () => {
try {
const granted = await hasCapability(
pluginId,
"network:serve",
manifest.capabilities,
);
if (!granted) {
res.status(403).json({
error: "Plugin is not granted the network:serve capability",
pluginId,
});
return;
}
next();
} catch (error) {
next(error);
}
})();
});
const authenticate = AuthManager.getInstance().createAuthMiddleware();
outer.use((req: Request, res: Response, next: NextFunction) => {
if (isPublicRequest(req, publicPaths)) {
pluginLogger.info(`Unauthenticated request to ${pluginId}${req.path}`, {
operation: "plugin_http_public_request",
});
next();
return;
}
authenticate(req, res, next);
});
if (!options?.rawBody) {
const limit = options?.bodyLimit ?? DEFAULT_BODY_LIMIT;
// Both parsers ignore a body they do not claim, so multipart still reaches
// multer untouched. database.ts skips its own global parser for
// /plugin-api so this limit is the one that actually applies.
outer.use(express.json({ limit }));
outer.use(express.urlencoded({ limit, extended: true }));
}
// After auth, so the actor is the user core authenticated rather than
// whatever the request claimed.
outer.use((req: Request, _res: Response, next: NextFunction) => {
const userId = (req as Request & { userId?: string }).userId;
if (!userId) {
next();
return;
}
runAsActor(userId, "request", () => next());
});
outer.use(inner);
outer.use(
(
error: unknown,
req: Request,
res: Response,
_next: NextFunction,
): void => {
void _next;
reportError(error);
pluginLogger.error(
`Plugin ${pluginId} route ${req.method} ${req.path} failed`,
error instanceof Error ? error : new Error(String(error)),
{ operation: "plugin_http_error" },
);
if (res.headersSent) {
// Mid-stream: the only honest thing left is to end it.
res.end();
return;
}
res.status(500).json({ error: "Plugin request failed" });
},
);
routers.set(pluginId, outer);
pluginLogger.info(`Mounted /plugin-api/${pluginId}`, {
operation: "plugin_http_mount",
});
return inner;
}
async function writePublicRouteAudit(
manifest: PluginManifest,
paths: string[],
): Promise<void> {
try {
const { logAudit } = await import("../utils/audit-logger.js");
await logAudit({
userId: "system",
username: `plugin:${manifest.id}`,
action: "plugin_http_public_routes",
resourceType: "plugin",
resourceId: manifest.id,
resourceName: manifest.name,
details: `unauthenticated routes: ${paths.join(", ")}`,
success: true,
});
} catch {
// Auditing must never stop a plugin starting.
}
}
/**
* Builds ctx.rbac.require.
*
* A plugin may only gate on a permission it declares itself. Without that a
* route could require admin.users.manage and borrow someone else's authority,
* which is the escalation shape the manifest validator already blocks for
* role defaults.
*/
export function createRbacMiddleware(
manifest: PluginManifest,
permission: string,
): PluginMiddleware {
const declared = new Set(
manifest.contributes?.permissionGroup?.permissions ?? [],
);
if (!declared.has(permission)) {
throw new Error(
`Plugin ${manifest.id} cannot require permission "${permission}": it is not declared in contributes.permissionGroup.permissions`,
);
}
return ((req: Request, res: Response, next: NextFunction) => {
void (async () => {
const userId = (req as Request & { userId?: string }).userId;
if (!userId) {
res.status(401).json({ error: "Authentication required" });
return;
}
try {
const { PermissionManager } =
await import("../utils/permission-manager.js");
const allowed = await PermissionManager.getInstance().hasPermission(
userId,
permission,
);
if (!allowed) {
res.status(403).json({ error: "Permission denied", permission });
return;
}
next();
} catch (error) {
next(error);
}
})();
}) as unknown as PluginMiddleware;
}
+9 -3
View File
@@ -10,7 +10,7 @@
*/
import { pluginLogger } from "../utils/logger.js";
import { unregisterPluginRouter } from "../database/routes/plugin-api-routes.js";
import { setPluginEnabledCheck, unregisterPluginHttp } from "./http.js";
import { PluginLoader, type LoadedPlugin } from "./loader.js";
import { invalidatePluginPermissionCache } from "./permissions.js";
@@ -19,6 +19,12 @@ let loader: PluginLoader | null = null;
export function getPluginRuntime(): { loader: PluginLoader } {
if (!loader) {
loader = new PluginLoader();
// The 503-while-disabled answer comes from here rather than from the
// router being torn down, so a request that arrives mid-disable gets a
// truthful status instead of a 404.
setPluginEnabledCheck(
(pluginId) => loader?.get(pluginId)?.state === "active",
);
}
return { loader };
}
@@ -336,7 +342,7 @@ export async function deactivatePlugin(pluginId: string): Promise<void> {
const plugin = pluginLoader.get(pluginId);
await pluginLoader.deactivate(pluginId);
unregisterPluginRouter(pluginId);
unregisterPluginHttp(pluginId);
const group = plugin?.manifest.contributes?.permissionGroup;
if (group) {
@@ -348,7 +354,7 @@ export async function deactivatePlugin(pluginId: string): Promise<void> {
export async function shutdownPlugins(): Promise<void> {
if (!loader) return;
for (const plugin of loader.list()) unregisterPluginRouter(plugin.id);
for (const plugin of loader.list()) unregisterPluginHttp(plugin.id);
await loader.shutdown();
}
+317
View File
@@ -0,0 +1,317 @@
/**
* WebSockets for plugins: one upgrade handler, no ports.
*
* A plugin registers /plugin-ws/<id>/<path> and core serves it from the main
* server's upgrade event. Auth is the same as every other socket in Termix
* (utils/ws-auth.ts), the actor is set for the handler, and every socket a
* plugin opened is closed when it deactivates.
*
* Two shapes, because two kinds of caller exist:
*
* - route(): core owns the WebSocketServer and hands the plugin a live
* socket. What a plugin writing its own protocol wants.
* - upgrade(): core hands over the raw upgrade after authenticating it, for
* a library that insists on owning its own server. guacamole-lite is the
* reason this exists: it constructs a WebSocketServer internally, so the
* only way to attach it is to give it the upgrade.
*
* Both run auth first, so "bring your own server" never means "bring your own
* auth".
*/
import type { IncomingMessage, Server as HttpServer } from "node:http";
import type { Duplex } from "node:stream";
import { WebSocketServer, type WebSocket } from "ws";
import type {
PluginWebSocketHandler,
PluginWebSocketOptions,
} from "@termix/plugin-sdk/backend";
import { pluginLogger } from "../utils/logger.js";
import { extractWebSocketToken } from "../utils/ws-auth.js";
import { runAsActor } from "./actor.js";
export const PLUGIN_WS_PREFIX = "/plugin-ws/";
type RawUpgradeHandler = (
request: IncomingMessage,
socket: Duplex,
head: Buffer,
userId: string,
) => void;
interface Route {
pluginId: string;
path: string;
options: PluginWebSocketOptions;
/** Declared capabilities, for the per-upgrade network:serve check. */
declared: readonly string[];
handler?: PluginWebSocketHandler;
rawHandler?: RawUpgradeHandler;
}
/** Keyed by "<pluginId>:<path>". */
const routes = new Map<string, Route>();
/** One server per plugin route, so disposal can close exactly its sockets. */
const servers = new Map<string, WebSocketServer>();
const attachedServers = new Set<HttpServer>();
function normalizePath(path: string): string {
const withSlash = path.startsWith("/") ? path : `/${path}`;
return withSlash.length > 1 && withSlash.endsWith("/")
? withSlash.slice(0, -1)
: withSlash;
}
function key(pluginId: string, path: string): string {
return `${pluginId}:${normalizePath(path)}`;
}
/**
* Parses /plugin-ws/<id>/<path> out of an upgrade URL.
*
* Returns null for anything else, so core's own sockets and a stray request
* both fall through to whatever else is listening rather than being claimed.
*/
export function parsePluginWsUrl(
url: string | undefined,
): { pluginId: string; path: string } | null {
if (!url) return null;
const pathname = url.split("?")[0];
if (!pathname.startsWith(PLUGIN_WS_PREFIX)) return null;
const rest = pathname.slice(PLUGIN_WS_PREFIX.length);
const slash = rest.indexOf("/");
if (slash <= 0) return null;
const pluginId = rest.slice(0, slash);
const path = normalizePath(rest.slice(slash));
if (!pluginId || path === "/") return null;
return { pluginId, path };
}
/**
* Resolves the user behind an upgrade, or null.
*
* A half-authenticated session is not a user: a token still awaiting TOTP is
* refused here exactly as createAuthMiddleware refuses it for HTTP, so a
* socket cannot be the way around the second factor.
*/
async function verifyToken(request: IncomingMessage): Promise<string | null> {
const token = extractWebSocketToken(request);
if (!token) return null;
try {
const { AuthManager } = await import("../utils/auth-manager.js");
const payload = await AuthManager.getInstance().verifyJWTToken(token);
if (!payload || payload.pendingTOTP) return null;
return payload.userId ?? null;
} catch {
return null;
}
}
function reject(socket: Duplex, status: number, message: string): void {
try {
socket.write(`HTTP/1.1 ${status} ${message}\r\nConnection: close\r\n\r\n`);
} catch {
// The peer may already be gone.
}
socket.destroy();
}
/**
* Serves a plugin upgrade, or returns false so the caller can pass it on.
*
* Exported for the test suite and for database.ts, which wires it into both
* the HTTP and the HTTPS server.
*/
export async function handlePluginUpgrade(
request: IncomingMessage,
socket: Duplex,
head: Buffer,
): Promise<boolean> {
const parsed = parsePluginWsUrl(request.url);
if (!parsed) return false;
const route = routes.get(key(parsed.pluginId, parsed.path));
if (!route) {
// Claimed by the prefix but nothing is listening: answering here is more
// honest than leaving the socket hanging for a plugin that is disabled.
reject(socket, 404, "Not Found");
return true;
}
// Same reasoning as the HTTP side: the grant is in the database, so it is
// checked per upgrade rather than at registration, and a revoke takes effect
// on the next connection without a restart.
const { hasCapability } = await import("./permissions.js");
if (!(await hasCapability(route.pluginId, "network:serve", route.declared))) {
reject(socket, 403, "Forbidden");
return true;
}
let userId = "";
if (!route.options.public) {
const resolved = await verifyToken(request);
if (!resolved) {
reject(socket, 401, "Unauthorized");
return true;
}
userId = resolved;
}
if (route.rawHandler) {
const rawHandler = route.rawHandler;
runAsActor(userId || "anonymous", "request", () =>
rawHandler(request, socket, head, userId),
);
return true;
}
const server = servers.get(key(route.pluginId, route.path));
if (!server) {
reject(socket, 503, "Service Unavailable");
return true;
}
server.handleUpgrade(request, socket, head, (ws: WebSocket) => {
const handler = route.handler;
if (!handler) {
ws.close(1011, "No handler");
return;
}
runAsActor(userId || "anonymous", "request", () => {
void Promise.resolve(handler({ userId, request, socket: ws })).catch(
(error) => {
pluginLogger.error(
`Plugin ${route.pluginId} socket ${route.path} failed`,
error instanceof Error ? error : new Error(String(error)),
{ operation: "plugin_ws_error" },
);
try {
ws.close(1011, "Plugin error");
} catch {
// Already closed.
}
},
);
});
});
return true;
}
/** Wires the upgrade handler onto a server. Idempotent per server. */
export function attachPluginWebSockets(server: HttpServer): void {
if (attachedServers.has(server)) return;
attachedServers.add(server);
server.on("upgrade", (request, socket, head) => {
void handlePluginUpgrade(
request as IncomingMessage,
socket as Duplex,
head as Buffer,
).catch((error) => {
pluginLogger.error(
"Plugin WebSocket upgrade failed",
error instanceof Error ? error : new Error(String(error)),
{ operation: "plugin_ws_upgrade" },
);
reject(socket as Duplex, 500, "Internal Server Error");
});
});
}
export function registerPluginWsRoute(
pluginId: string,
path: string,
handler: PluginWebSocketHandler,
declared: readonly string[],
options: PluginWebSocketOptions = {},
): () => void {
const routeKey = key(pluginId, path);
const normalized = normalizePath(path);
routes.set(routeKey, {
pluginId,
path: normalized,
options,
declared,
handler,
});
servers.set(routeKey, new WebSocketServer({ noServer: true }));
pluginLogger.info(`Mounted /plugin-ws/${pluginId}${normalized}`, {
operation: "plugin_ws_mount",
});
return () => disposeRoute(routeKey, pluginId, normalized);
}
export function registerPluginWsUpgrade(
pluginId: string,
path: string,
handler: RawUpgradeHandler,
declared: readonly string[],
options: PluginWebSocketOptions = {},
): () => void {
const routeKey = key(pluginId, path);
const normalized = normalizePath(path);
routes.set(routeKey, {
pluginId,
path: normalized,
options,
declared,
rawHandler: handler,
});
pluginLogger.info(`Mounted raw upgrade /plugin-ws/${pluginId}${normalized}`, {
operation: "plugin_ws_mount",
});
return () => disposeRoute(routeKey, pluginId, normalized);
}
function disposeRoute(routeKey: string, pluginId: string, path: string): void {
routes.delete(routeKey);
const server = servers.get(routeKey);
if (server) {
servers.delete(routeKey);
// Close the clients first: server.close() waits for them to drain, and a
// live terminal would otherwise keep the route alive after deactivate.
for (const client of server.clients) {
try {
client.close(1001, "Plugin disabled");
} catch {
// Already gone.
}
}
try {
server.close();
} catch {
// Closing twice is not an error worth surfacing.
}
}
pluginLogger.info(`Unmounted /plugin-ws/${pluginId}${path}`, {
operation: "plugin_ws_unmount",
});
}
export function getRegisteredWsRoutes(): string[] {
return [...routes.keys()];
}
/** Test seam. */
export function resetPluginWebSockets(): void {
for (const routeKey of [...routes.keys()]) {
const route = routes.get(routeKey);
if (route) disposeRoute(routeKey, route.pluginId, route.path);
}
routes.clear();
servers.clear();
attachedServers.clear();
}