mirror of
https://github.com/Termix-SSH/Termix.git
synced 2026-10-29 14:45:24 +00:00
feat: harden plugin api for separate repos, signed updates and generic external login
This commit is contained in:
1 parent
fb9c168b6b
commit
80944301dd
310 files changed
+6577
-4927
No files matched your search
@@ -46,15 +46,3 @@ export function getActorSessionId(): string | undefined {
|
||||
export function getActor(): string | undefined {
|
||||
return storage.getStore()?.userId;
|
||||
}
|
||||
|
||||
class PluginActorError extends Error {
|
||||
readonly code = "EPLUGINACTOR";
|
||||
|
||||
constructor(pluginId: string, operation: string) {
|
||||
super(
|
||||
`Plugin "${pluginId}" called ${operation} with no acting user. ` +
|
||||
`Wrap background work in ctx.asUser(userId, fn).`,
|
||||
);
|
||||
this.name = "PluginActorError";
|
||||
}
|
||||
}
|
||||
@@ -93,12 +93,7 @@ export function describePluginFrontend(
|
||||
const entry = frontendEntry(plugin);
|
||||
if (!entry || !fileExists(entry.file)) return { ...none, locales };
|
||||
|
||||
const stat = fs.statSync(entry.file);
|
||||
const assetVersion = crypto
|
||||
.createHash("sha256")
|
||||
.update(`${plugin.manifest.version}:${stat.size}:${stat.mtimeMs}`)
|
||||
.digest("hex")
|
||||
.slice(0, 12);
|
||||
const assetVersion = contentVersion(entry.file, plugin.manifest.version);
|
||||
|
||||
return {
|
||||
frontend: true,
|
||||
@@ -108,6 +103,29 @@ export function describePluginFrontend(
|
||||
};
|
||||
}
|
||||
|
||||
const versionCache = new Map<string, { stamp: string; version: string }>();
|
||||
|
||||
/**
|
||||
* A hash of the bundle's bytes, not its mtime: assets are cached as immutable
|
||||
* for a year, and a rebuilt .tmxplug has a fixed mtime, so a same-size rebuild
|
||||
* would otherwise keep serving the stale copy.
|
||||
*/
|
||||
function contentVersion(file: string, version: string): string {
|
||||
const stat = fs.statSync(file);
|
||||
// ctime, unlike mtime, cannot be set by tar, so an unpack always changes it.
|
||||
const stamp = `${version}:${stat.size}:${stat.mtimeMs}:${stat.ctimeMs}`;
|
||||
const cached = versionCache.get(file);
|
||||
if (cached?.stamp === stamp) return cached.version;
|
||||
const hashed = crypto
|
||||
.createHash("sha256")
|
||||
.update(version)
|
||||
.update(fs.readFileSync(file))
|
||||
.digest("hex")
|
||||
.slice(0, 12);
|
||||
versionCache.set(file, { stamp, version: hashed });
|
||||
return hashed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps a request path under /plugin-assets/<id>/ to a file, or null.
|
||||
*
|
||||
|
||||
@@ -81,7 +81,6 @@ export const HOST_SECRET_FIELDS = [
|
||||
"passphrase",
|
||||
"sudoPassword",
|
||||
"socks5Password",
|
||||
"vaultToken",
|
||||
] as const;
|
||||
|
||||
/** A copy of a host with every secret field removed, nested ones included. */
|
||||
|
||||
@@ -51,6 +51,34 @@ async function sha256Of(file: string): Promise<string | null> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Downloads through the SSRF guard, following https redirects by hand
|
||||
* because the guard refuses automatic ones (release assets redirect to a
|
||||
* CDN).
|
||||
*/
|
||||
async function guardedDownload(
|
||||
url: URL | string,
|
||||
init: RequestInit,
|
||||
): Promise<Response> {
|
||||
const { safeOutboundFetch } = await import("../utils/safe-outbound-fetch.js");
|
||||
let current = new URL(url.toString());
|
||||
for (let hop = 0; hop < 5; hop++) {
|
||||
const response = await safeOutboundFetch(current.toString(), {
|
||||
...init,
|
||||
redirect: "manual",
|
||||
});
|
||||
const location = response.headers.get("location");
|
||||
if (response.status < 300 || response.status >= 400 || !location) {
|
||||
return response;
|
||||
}
|
||||
current = new URL(location, current);
|
||||
if (current.protocol !== "https:") {
|
||||
throw new Error("Binary downloads must stay on https");
|
||||
}
|
||||
}
|
||||
throw new Error("Too many redirects");
|
||||
}
|
||||
|
||||
function errorMessage(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error);
|
||||
}
|
||||
@@ -60,7 +88,7 @@ export function createPluginProcess(deps: Deps): PluginProcess {
|
||||
const declared = deps.manifest.capabilities;
|
||||
const binDir =
|
||||
deps.binDir ?? (() => path.join(getPluginDataDir(pluginId), "bin"));
|
||||
const doFetch = deps.fetch ?? fetch;
|
||||
const doFetch = deps.fetch ?? (guardedDownload as typeof fetch);
|
||||
|
||||
const audited = async <T>(
|
||||
action: string,
|
||||
|
||||
@@ -791,11 +791,12 @@ export function createPluginAuth({ manifest, bag, audit }: Deps): PluginAuth {
|
||||
await granted();
|
||||
if (!match.sub && !match.sid) return 0;
|
||||
const { AuthManager } = await import("../utils/auth-manager.js");
|
||||
const revoked = await AuthManager.getInstance().revokeSessionsByOidc({
|
||||
ssoProviderId: match.providerId ?? null,
|
||||
sub: match.sub ?? null,
|
||||
sid: match.sid ?? null,
|
||||
});
|
||||
const revoked =
|
||||
await AuthManager.getInstance().revokeSessionsByExternalSession({
|
||||
ssoProviderId: match.providerId ?? null,
|
||||
sub: match.sub ?? null,
|
||||
sid: match.sid ?? null,
|
||||
});
|
||||
await audit(
|
||||
"auth_sessions_revoked",
|
||||
`${revoked} session(s) for provider ${match.providerId ?? "none"}`,
|
||||
|
||||
+83
-22
@@ -47,7 +47,7 @@ import {
|
||||
type PluginContext,
|
||||
type PluginModule,
|
||||
type PluginOpenIsolatedWindowRequest,
|
||||
type PluginNativeRdpRequest,
|
||||
type PluginExternalClientRequest,
|
||||
} from "@termix/plugin-sdk/backend";
|
||||
import type { PluginTableDefinition } from "@termix/plugin-sdk/db";
|
||||
import * as syncRegistry from "./sync-registry.js";
|
||||
@@ -191,14 +191,17 @@ export function createPluginContext(
|
||||
await import("../database/repositories/factory.js");
|
||||
return createCurrentPluginSettingsRepository();
|
||||
};
|
||||
// A user named by the plugin instead of the ambient actor. Honoured, like
|
||||
// ctx.asUser, and audited the same way.
|
||||
// A user named by the plugin instead of the ambient actor. The same thing
|
||||
// as ctx.asUser, so it needs the same capability and gets the same audit.
|
||||
const namedUser = (
|
||||
userId: string | undefined,
|
||||
via: string,
|
||||
): string | undefined => {
|
||||
if (!userId) return getActor();
|
||||
if (userId !== getActor()) {
|
||||
if (!declared.includes("users:impersonate")) {
|
||||
throw capabilityRefused(pluginId, "users:impersonate");
|
||||
}
|
||||
void writeAudit(
|
||||
manifest,
|
||||
{
|
||||
@@ -342,11 +345,11 @@ export function createPluginContext(
|
||||
|
||||
const dbRefs = guarded(
|
||||
manifest,
|
||||
"db:own",
|
||||
"db:core-refs",
|
||||
async () => {
|
||||
// Read-only by convention, not by engine: a plugin holding these can
|
||||
// write through them. The capability and the audit line are the record
|
||||
// that it did. See the header of this file.
|
||||
// write through them, which is why the capability is rated high. The
|
||||
// audit line is the record that it asked.
|
||||
const schema = await import("../database/db/schema.js");
|
||||
return {
|
||||
users: schema.users,
|
||||
@@ -380,25 +383,25 @@ export function createPluginContext(
|
||||
},
|
||||
);
|
||||
|
||||
const desktopLaunchNativeRdp = guarded(
|
||||
const desktopLaunchExternalClient = guarded(
|
||||
manifest,
|
||||
"desktop:window",
|
||||
async (request: PluginNativeRdpRequest) => {
|
||||
async (request: PluginExternalClientRequest) => {
|
||||
const { isElectronIpcAvailable, requestFromElectronMain } =
|
||||
await import("../utils/electron-ipc-bridge.js");
|
||||
if (!isElectronIpcAvailable()) {
|
||||
throw new Error(
|
||||
`Plugin ${pluginId} tried to open the native RDP client outside the desktop app`,
|
||||
`Plugin ${pluginId} tried to open an external client outside the desktop app`,
|
||||
);
|
||||
}
|
||||
return requestFromElectronMain<{ success: boolean; error?: string }>(
|
||||
"launch-native-rdp",
|
||||
"launch-external-client",
|
||||
request,
|
||||
);
|
||||
},
|
||||
{
|
||||
action: "desktop_launch_native_rdp",
|
||||
details: () => "opened the native RDP client",
|
||||
action: "desktop_launch_external_client",
|
||||
details: () => "opened an external client",
|
||||
},
|
||||
);
|
||||
|
||||
@@ -455,7 +458,26 @@ export function createPluginContext(
|
||||
`Declare the events:core capability to listen to core topics.`,
|
||||
);
|
||||
}
|
||||
const unsubscribe = pluginEvents.on(topic, listener);
|
||||
// A throwing listener counts against the plugin's error budget.
|
||||
const counted: typeof listener = (payload) => {
|
||||
try {
|
||||
const result = listener(payload) as unknown;
|
||||
if (
|
||||
result &&
|
||||
typeof (result as Promise<void>).then === "function"
|
||||
) {
|
||||
return (result as Promise<void>).catch((error) => {
|
||||
void reportRuntimeError(pluginId, error);
|
||||
throw error;
|
||||
});
|
||||
}
|
||||
return result as never;
|
||||
} catch (error) {
|
||||
void reportRuntimeError(pluginId, error);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
const unsubscribe = pluginEvents.on(topic, counted);
|
||||
handle.bag.add(unsubscribe, `event listener for "${topic}"`);
|
||||
return unsubscribe;
|
||||
},
|
||||
@@ -502,7 +524,10 @@ export function createPluginContext(
|
||||
},
|
||||
|
||||
registry: {
|
||||
// Keys are namespaced by plugin id, so one plugin cannot replace or
|
||||
// remove what another provides. Anyone may consume.
|
||||
provide: (key, value) => {
|
||||
assertOwnRegistryKey(pluginId, key);
|
||||
registry.provide(key, value);
|
||||
handle.bag.add(
|
||||
() => void registry.revoke(key, value),
|
||||
@@ -510,7 +535,10 @@ export function createPluginContext(
|
||||
);
|
||||
},
|
||||
consume: (key) => registry.consume(key),
|
||||
revoke: (key, value) => registry.revoke(key, value),
|
||||
revoke: (key, value) => {
|
||||
assertOwnRegistryKey(pluginId, key);
|
||||
return registry.revoke(key, value);
|
||||
},
|
||||
},
|
||||
|
||||
services: {
|
||||
@@ -548,8 +576,18 @@ export function createPluginContext(
|
||||
);
|
||||
},
|
||||
|
||||
get: (service, options) =>
|
||||
serviceRegistry.createServiceHandle(
|
||||
get: (service, options) => {
|
||||
// Declared, like getShared: a plugin reaches only the services its
|
||||
// manifest names in requires, or its own.
|
||||
const known =
|
||||
manifest.requires?.some((entry) => entry.service === service) ||
|
||||
manifest.provides?.some((entry) => entry.service === service);
|
||||
if (!known) {
|
||||
throw new Error(
|
||||
`Plugin ${pluginId} cannot use service "${service}": it is not declared in the manifest's requires array`,
|
||||
);
|
||||
}
|
||||
return serviceRegistry.createServiceHandle(
|
||||
service,
|
||||
pluginId,
|
||||
{
|
||||
@@ -562,7 +600,8 @@ export function createPluginContext(
|
||||
audit: (entry) => writeServiceAudit(entry),
|
||||
},
|
||||
options?.provider ?? "",
|
||||
),
|
||||
);
|
||||
},
|
||||
|
||||
providers: (service) => serviceRegistry.listProviderNames(service),
|
||||
},
|
||||
@@ -709,10 +748,20 @@ export function createPluginContext(
|
||||
if (!declared.includes("network:serve")) {
|
||||
throw capabilityRefused(pluginId, "network:serve");
|
||||
}
|
||||
// Only a signed-in caller's socket counts against the error budget,
|
||||
// so an anonymous peer cannot switch the plugin off.
|
||||
const counted: typeof wsHandler = async (connection) => {
|
||||
try {
|
||||
return await wsHandler(connection);
|
||||
} catch (error) {
|
||||
if (connection.userId) void reportRuntimeError(pluginId, error);
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
const dispose = registerPluginWsRoute(
|
||||
pluginId,
|
||||
path,
|
||||
wsHandler,
|
||||
counted,
|
||||
declared,
|
||||
options,
|
||||
);
|
||||
@@ -840,13 +889,14 @@ export function createPluginContext(
|
||||
|
||||
hosts: createPluginHosts({ manifest, bag: handle.bag, audit: auditCall }),
|
||||
|
||||
schedule: createPluginSchedule(handle.bag, (message, error) =>
|
||||
schedule: createPluginSchedule(handle.bag, (message, error) => {
|
||||
pluginLogger.error(
|
||||
message,
|
||||
error instanceof Error ? error : new Error(String(error)),
|
||||
logContext,
|
||||
),
|
||||
),
|
||||
);
|
||||
void reportRuntimeError(pluginId, error);
|
||||
}),
|
||||
ssh: createPluginSsh({ manifest, bag: handle.bag, audit: auditCall }),
|
||||
notify: createPluginNotify({
|
||||
manifest,
|
||||
@@ -864,7 +914,7 @@ export function createPluginContext(
|
||||
|
||||
desktop: {
|
||||
openIsolatedWindow: (request) => desktopOpenIsolatedWindow(request),
|
||||
launchNativeRdp: (request) => desktopLaunchNativeRdp(request),
|
||||
launchExternalClient: (request) => desktopLaunchExternalClient(request),
|
||||
available: () => isElectronIpcAvailable(),
|
||||
},
|
||||
|
||||
@@ -909,6 +959,9 @@ export function createPluginContext(
|
||||
`Plugin ${pluginId} called ctx.asUser without a user id`,
|
||||
);
|
||||
}
|
||||
if (!declared.includes("users:impersonate")) {
|
||||
throw capabilityRefused(pluginId, "users:impersonate");
|
||||
}
|
||||
await writeAudit(
|
||||
manifest,
|
||||
{
|
||||
@@ -942,6 +995,14 @@ async function reportRuntimeError(
|
||||
}
|
||||
}
|
||||
|
||||
function assertOwnRegistryKey(pluginId: string, key: string): void {
|
||||
if (typeof key !== "string" || !key.startsWith(`${pluginId}.`)) {
|
||||
throw new Error(
|
||||
`Plugin ${pluginId} may only use registry keys under "${pluginId}.", not "${key}"`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function assertKvKey(key: string): void {
|
||||
if (typeof key !== "string" || key.length === 0) {
|
||||
throw new Error("ctx.kv keys must be non-empty strings");
|
||||
|
||||
+75
-11
@@ -30,14 +30,19 @@ export function registerTable(
|
||||
definition: PluginTableDefinition,
|
||||
): unknown {
|
||||
const definitions = registered.get(pluginId) ?? [];
|
||||
const existing = definitions.find((entry) => entry.name === definition.name);
|
||||
if (existing) {
|
||||
// Re-registering on re-activation is normal: the module is cached, so
|
||||
// activate runs again against the same definitions.
|
||||
return getTable(pluginId, definition.name);
|
||||
const index = definitions.findIndex(
|
||||
(entry) => entry.name === definition.name,
|
||||
);
|
||||
if (index >= 0) {
|
||||
// Re-registering on re-activation is normal. The same definition keeps
|
||||
// its table object; a changed one (an updated plugin) is rebuilt.
|
||||
if (JSON.stringify(definitions[index]) === JSON.stringify(definition)) {
|
||||
return getTable(pluginId, definition.name);
|
||||
}
|
||||
definitions[index] = definition;
|
||||
} else {
|
||||
definitions.push(definition);
|
||||
}
|
||||
|
||||
definitions.push(definition);
|
||||
registered.set(pluginId, definitions);
|
||||
|
||||
const table = buildTable(pluginId, definition);
|
||||
@@ -86,12 +91,65 @@ async function execute(statements: string[]): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Postgres runs DDL inside a transaction, so one migration file applies
|
||||
* whole or not at all. Through drizzle's transaction because a pool would
|
||||
* spread a raw BEGIN and the statements over different connections.
|
||||
*/
|
||||
async function executeInTransaction(statements: string[]): Promise<void> {
|
||||
const { getDb } = await import("../database/db/index.js");
|
||||
const db = getDb() as unknown as {
|
||||
transaction: (
|
||||
fn: (tx: {
|
||||
execute: (query: unknown) => Promise<unknown>;
|
||||
}) => Promise<void>,
|
||||
) => Promise<void>;
|
||||
};
|
||||
await db.transaction(async (tx) => {
|
||||
for (const statement of statements) {
|
||||
await tx.execute(sql.raw(statement));
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* SQLite has one connection, so a raw BEGIN covers the statements and the
|
||||
* migration record together.
|
||||
*/
|
||||
async function sqliteTransaction(fn: () => Promise<void>): Promise<void> {
|
||||
const { runStatement } =
|
||||
await import("../utils/crypto-migration/raw-rows.js");
|
||||
await runStatement(sql.raw("BEGIN"));
|
||||
try {
|
||||
await fn();
|
||||
await runStatement(sql.raw("COMMIT"));
|
||||
} catch (error) {
|
||||
await runStatement(sql.raw("ROLLBACK")).catch(() => {});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
async function executeNamingFailure(statements: string[]): Promise<void> {
|
||||
const { runStatement } =
|
||||
await import("../utils/crypto-migration/raw-rows.js");
|
||||
for (const [index, statement] of statements.entries()) {
|
||||
try {
|
||||
await runStatement(sql.raw(statement));
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
throw new Error(
|
||||
`statement ${index + 1} of ${statements.length} failed (earlier ones are already applied on MySQL): ${message}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the runner that applies migrations against the live database.
|
||||
*
|
||||
* No transaction on any engine: MySQL commits implicitly on DDL, and each
|
||||
* migration is recorded once it has run, so a retry after a failure starts
|
||||
* at the migration that failed.
|
||||
* SQLite and Postgres apply each migration file atomically. MySQL commits
|
||||
* implicitly on DDL, so there a failed file can leave earlier statements
|
||||
* applied; the error names the statement so it can be fixed by hand.
|
||||
*/
|
||||
async function createMigrationRunner(
|
||||
dialect: DatabaseDialect = resolveDatabaseDialect(),
|
||||
@@ -109,7 +167,13 @@ async function createMigrationRunner(
|
||||
checksum: row.checksum,
|
||||
}));
|
||||
},
|
||||
execute,
|
||||
execute:
|
||||
dialect === "postgres"
|
||||
? executeInTransaction
|
||||
: dialect === "mysql"
|
||||
? executeNamingFailure
|
||||
: execute,
|
||||
transaction: dialect === "sqlite" ? sqliteTransaction : undefined,
|
||||
record: async (pluginId, migration) => {
|
||||
await repository.record(pluginId, migration.id, migration.checksum);
|
||||
},
|
||||
|
||||
@@ -38,6 +38,43 @@ const DEFAULT_BODY_LIMIT = "2mb";
|
||||
/** Routers by plugin id, read by the dispatcher in plugin-api-routes.ts. */
|
||||
const routers = new Map<string, Router>();
|
||||
|
||||
/**
|
||||
* Requests per minute one address may make to one plugin's public routes.
|
||||
* They run without a login, so they are the cheap way to hammer a plugin.
|
||||
*/
|
||||
function publicRateLimit(): number {
|
||||
const parsed = Number(process.env.TERMIX_PLUGIN_PUBLIC_RATE_LIMIT);
|
||||
return Number.isFinite(parsed) && parsed > 0 ? parsed : 120;
|
||||
}
|
||||
const PUBLIC_RATE_WINDOW_MS = 60_000;
|
||||
const publicHits = new Map<string, { count: number; resetAt: number }>();
|
||||
|
||||
/** True when this request is over the public route limit. */
|
||||
export function overPublicRateLimit(
|
||||
pluginId: string,
|
||||
ip: string,
|
||||
now = Date.now(),
|
||||
): boolean {
|
||||
if (publicHits.size > 10_000) {
|
||||
for (const [hitKey, entry] of publicHits) {
|
||||
if (entry.resetAt <= now) publicHits.delete(hitKey);
|
||||
}
|
||||
}
|
||||
const hitKey = `${pluginId}|${ip}`;
|
||||
const entry = publicHits.get(hitKey);
|
||||
if (!entry || entry.resetAt <= now) {
|
||||
publicHits.set(hitKey, { count: 1, resetAt: now + PUBLIC_RATE_WINDOW_MS });
|
||||
return false;
|
||||
}
|
||||
entry.count++;
|
||||
return entry.count > publicRateLimit();
|
||||
}
|
||||
|
||||
/** Test seam. */
|
||||
export function resetPublicRateLimits(): void {
|
||||
publicHits.clear();
|
||||
}
|
||||
|
||||
/** Declared public paths by plugin id, for the admin plugin details. */
|
||||
const publicRoutes = new Map<string, string[]>();
|
||||
|
||||
@@ -151,6 +188,11 @@ export function createPluginRouter({
|
||||
reportError,
|
||||
}: CreatePluginRouterArgs): Router {
|
||||
const pluginId = manifest.id;
|
||||
if (routers.has(pluginId)) {
|
||||
throw new Error(
|
||||
`Plugin ${pluginId} already has a router; call ctx.http.router() once`,
|
||||
);
|
||||
}
|
||||
const declaredPublic = (options?.public ?? []).map((path) =>
|
||||
normalizePublicPath(path),
|
||||
);
|
||||
@@ -209,6 +251,11 @@ export function createPluginRouter({
|
||||
|
||||
outer.use((req: Request, res: Response, next: NextFunction) => {
|
||||
if (isPublicRequest(req, publicPaths)) {
|
||||
const ip = req.ip || req.socket?.remoteAddress || "unknown";
|
||||
if (overPublicRateLimit(pluginId, ip)) {
|
||||
res.status(429).json({ error: "Too many requests" });
|
||||
return;
|
||||
}
|
||||
pluginLogger.info(`Unauthenticated request to ${pluginId}${req.path}`, {
|
||||
operation: "plugin_http_public_request",
|
||||
});
|
||||
|
||||
@@ -29,7 +29,9 @@ let loader: PluginLoader | null = null;
|
||||
|
||||
export function getPluginRuntime(): { loader: PluginLoader } {
|
||||
if (!loader) {
|
||||
loader = new PluginLoader();
|
||||
loader = new PluginLoader({
|
||||
onFailed: (plugin) => void handlePluginFailed(plugin),
|
||||
});
|
||||
// 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.
|
||||
@@ -91,6 +93,29 @@ export function getPluginRuntime(): { loader: PluginLoader } {
|
||||
return { loader };
|
||||
}
|
||||
|
||||
/**
|
||||
* A plugin that tripped the error budget is off until an admin retries it, so
|
||||
* it has to look off everywhere: no routes, greyed permissions, and a failed
|
||||
* row that survives a restart.
|
||||
*/
|
||||
async function handlePluginFailed(plugin: LoadedPlugin): Promise<void> {
|
||||
unregisterPluginHttp(plugin.id);
|
||||
try {
|
||||
if (plugin.manifest.contributes?.permissions?.length) {
|
||||
const { markPluginPermissionsDisabled } =
|
||||
await import("../utils/permission-catalog.js");
|
||||
markPluginPermissionsDisabled(plugin.id);
|
||||
}
|
||||
await persistRuntimeState([plugin]);
|
||||
} catch (error) {
|
||||
pluginLogger.error(
|
||||
`Could not record that plugin ${plugin.id} failed`,
|
||||
error instanceof Error ? error : new Error(String(error)),
|
||||
{ operation: "plugin_runtime" },
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Gives every discovered plugin a row, and refreshes the stored manifest.
|
||||
*
|
||||
@@ -533,13 +558,10 @@ export async function installPluginArtifact(
|
||||
file: string,
|
||||
): Promise<LoadedPlugin> {
|
||||
const { loader: pluginLoader } = getPluginRuntime();
|
||||
const bundledIds = new Set(
|
||||
pluginLoader
|
||||
.list()
|
||||
.filter((plugin) => plugin.source === "bundled")
|
||||
.map((plugin) => plugin.id),
|
||||
const plugin = await pluginLoader.loadArtifact(
|
||||
file,
|
||||
pluginLoader.bundledIds(),
|
||||
);
|
||||
const plugin = await pluginLoader.loadArtifact(file, bundledIds);
|
||||
await seedPlugins([plugin]);
|
||||
await syncCapabilityGrants([plugin]);
|
||||
return plugin;
|
||||
|
||||
+129
-37
@@ -18,7 +18,8 @@ import path from "node:path";
|
||||
import semver from "semver";
|
||||
import * as tar from "tar";
|
||||
import { pluginLogger } from "../utils/logger.js";
|
||||
import { parseManifest } from "./manifest.js";
|
||||
import { isTermixCompatible, parseManifest } from "./manifest.js";
|
||||
import { getLocalVersion } from "../utils/app-version.js";
|
||||
import type { PluginManifest } from "@termix/plugin-sdk/manifest";
|
||||
import {
|
||||
getBundledPluginsDir,
|
||||
@@ -67,6 +68,13 @@ export interface LoadedPlugin {
|
||||
lastError: string | null;
|
||||
/** Timestamps of recent runtime errors, trimmed to the window. */
|
||||
errorTimestamps: number[];
|
||||
/**
|
||||
* The .tmxplug this copy was unpacked from. Set for user plugins and for a
|
||||
* signed update that replaced a bundled plugin.
|
||||
*/
|
||||
artifact?: string;
|
||||
/** Key id that signed the artifact, when a trusted key signed it. */
|
||||
signedBy?: string;
|
||||
}
|
||||
|
||||
export interface PluginLoaderOptions {
|
||||
@@ -78,6 +86,8 @@ export interface PluginLoaderOptions {
|
||||
|
||||
export class PluginLoader {
|
||||
private readonly plugins = new Map<string, LoadedPlugin>();
|
||||
/** Version of each bundled plugin as shipped, the floor for any update. */
|
||||
private readonly bundledVersions = new Map<string, string>();
|
||||
/** Activation order, so shutdown can run it backwards. */
|
||||
private activationOrder: string[] = [];
|
||||
|
||||
@@ -91,6 +101,16 @@ export class PluginLoader {
|
||||
return this.plugins.get(pluginId);
|
||||
}
|
||||
|
||||
/** Ids of the plugins that ship with this install, loaded or replaced. */
|
||||
bundledIds(): Set<string> {
|
||||
return new Set([
|
||||
...this.bundledVersions.keys(),
|
||||
...this.list()
|
||||
.filter((plugin) => plugin.source === "bundled")
|
||||
.map((plugin) => plugin.id),
|
||||
]);
|
||||
}
|
||||
|
||||
/** Drops a stopped plugin from the list, so a new copy can load in its place. */
|
||||
forget(pluginId: string): void {
|
||||
const plugin = this.plugins.get(pluginId);
|
||||
@@ -122,6 +142,12 @@ export class PluginLoader {
|
||||
);
|
||||
}
|
||||
|
||||
if (!isTermixCompatible(manifest.engine.termix, getLocalVersion())) {
|
||||
throw new Error(
|
||||
`Plugin ${manifest.id} needs Termix ${manifest.engine.termix}, but this is ${getLocalVersion()}`,
|
||||
);
|
||||
}
|
||||
|
||||
if (path.basename(dir) !== manifest.id) {
|
||||
throw new Error(
|
||||
`Plugin directory "${path.basename(dir)}" does not match manifest id "${manifest.id}"`,
|
||||
@@ -141,6 +167,10 @@ export class PluginLoader {
|
||||
);
|
||||
}
|
||||
|
||||
if (this.plugins.has(manifest.id)) {
|
||||
throw new Error(`another plugin already uses the id "${manifest.id}"`);
|
||||
}
|
||||
|
||||
const overlap = [...this.plugins.values()].find((other) =>
|
||||
prefixesOverlap(other.id, manifest.id),
|
||||
);
|
||||
@@ -175,16 +205,16 @@ export class PluginLoader {
|
||||
}
|
||||
|
||||
/**
|
||||
* Scans both roots. A user-installed plugin can never shadow a bundled one:
|
||||
* the collision is rejected with an error rather than silently skipped, so
|
||||
* dropping a folder with a bundled plugin's id into the data directory is visibly
|
||||
* refused instead of quietly ignored.
|
||||
* Scans both roots. A user plugin folder can never shadow a bundled one: the
|
||||
* collision is rejected with an error rather than silently skipped.
|
||||
*
|
||||
* The check is on the directory name, which load() then requires to equal
|
||||
* the manifest id, and it covers bundled folders that failed to load too.
|
||||
* The one way to replace a bundled plugin is a .tmxplug signed by a trusted
|
||||
* key with a higher version (see loadArtifact). That is how one official
|
||||
* plugin ships an update without a core release.
|
||||
*/
|
||||
async loadAll(): Promise<LoadedPlugin[]> {
|
||||
this.plugins.clear();
|
||||
this.bundledVersions.clear();
|
||||
const loaded: LoadedPlugin[] = [];
|
||||
// Every bundled folder name, loaded or not: a bundled plugin that fails
|
||||
// to load must not leave its id free for a user plugin to take.
|
||||
@@ -216,13 +246,16 @@ export class PluginLoader {
|
||||
const dir = path.join(root, entry.name);
|
||||
try {
|
||||
if (isArtifact) {
|
||||
loaded.push(await this.loadArtifact(dir, bundledIds));
|
||||
const plugin = await this.loadArtifact(dir, bundledIds);
|
||||
const index = loaded.findIndex((other) => other.id === plugin.id);
|
||||
if (index >= 0) loaded[index] = plugin;
|
||||
else loaded.push(plugin);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (source === "user" && bundledIds.has(entry.name)) {
|
||||
throw new Error(
|
||||
`a bundled plugin already uses the id "${entry.name}"; a plugin in the data directory cannot replace it`,
|
||||
`a bundled plugin already uses the id "${entry.name}"; only a signed .tmxplug with a higher version can replace it`,
|
||||
);
|
||||
}
|
||||
|
||||
@@ -236,6 +269,9 @@ export class PluginLoader {
|
||||
}
|
||||
|
||||
const plugin = await this.load(dir, source);
|
||||
if (source === "bundled") {
|
||||
this.bundledVersions.set(plugin.id, plugin.manifest.version);
|
||||
}
|
||||
loaded.push(plugin);
|
||||
} catch (error) {
|
||||
pluginLogger.error(
|
||||
@@ -255,6 +291,10 @@ export class PluginLoader {
|
||||
* against a pinned key; with TERMIX_REQUIRE_SIGNED_PLUGINS=true it must
|
||||
* also exist. The archive is unpacked fresh on every load, so the file
|
||||
* stays the only source of truth.
|
||||
*
|
||||
* An artifact with a bundled plugin's id is an update to it. It replaces
|
||||
* the bundled copy only when a trusted key signed it and its version is
|
||||
* higher; if it then fails to load the bundled copy stays.
|
||||
*/
|
||||
async loadArtifact(
|
||||
file: string,
|
||||
@@ -262,6 +302,7 @@ export class PluginLoader {
|
||||
): Promise<LoadedPlugin> {
|
||||
const buffer = await fs.promises.readFile(file);
|
||||
const sigFile = `${file}.sig`;
|
||||
let signedBy: string | undefined;
|
||||
|
||||
if (fs.existsSync(sigFile)) {
|
||||
const sha256 = crypto.createHash("sha256").update(buffer).digest("hex");
|
||||
@@ -273,6 +314,7 @@ export class PluginLoader {
|
||||
if (result.ok === false) {
|
||||
throw new Error(`signature check failed: ${result.reason}`);
|
||||
}
|
||||
signedBy = result.keyId;
|
||||
} else if (requireSignedPlugins()) {
|
||||
throw new Error(
|
||||
`TERMIX_REQUIRE_SIGNED_PLUGINS is on and ${path.basename(file)} has no .sig next to it`,
|
||||
@@ -301,22 +343,47 @@ export class PluginLoader {
|
||||
await fs.promises.readFile(getPluginManifestPath(staging), "utf8"),
|
||||
);
|
||||
const id = typeof raw?.id === "string" ? raw.id : "";
|
||||
if (!/^[a-z0-9][a-z0-9-]*$/.test(id)) {
|
||||
if (!PLUGIN_ID_PATTERN.test(id)) {
|
||||
throw new Error(`${path.basename(file)} has no valid manifest id`);
|
||||
}
|
||||
if (bundledIds.has(id)) {
|
||||
throw new Error(
|
||||
`a bundled plugin already uses the id "${id}"; a plugin in the data directory cannot replace it`,
|
||||
|
||||
const isUpdate = bundledIds.has(id);
|
||||
const replaced = isUpdate ? this.plugins.get(id) : undefined;
|
||||
if (isUpdate) {
|
||||
assertBundledUpdate(
|
||||
id,
|
||||
raw?.version,
|
||||
signedBy,
|
||||
this.bundledVersions.get(id) ?? replaced?.manifest.version,
|
||||
);
|
||||
}
|
||||
if (this.plugins.has(id)) {
|
||||
} else if (this.plugins.has(id)) {
|
||||
throw new Error(`another plugin already uses the id "${id}"`);
|
||||
}
|
||||
if (replaced && replaced.state !== "loaded") {
|
||||
throw new Error(`plugin ${id} is running; stop it before updating`);
|
||||
}
|
||||
|
||||
const target = path.join(unpackedRoot, id);
|
||||
await fs.promises.rm(target, { recursive: true, force: true });
|
||||
await fs.promises.rename(staging, target);
|
||||
return await this.load(target, "user");
|
||||
|
||||
if (replaced) this.plugins.delete(id);
|
||||
try {
|
||||
// A signed update keeps the bundled trust it replaces.
|
||||
const plugin = await this.load(target, isUpdate ? "bundled" : "user");
|
||||
plugin.artifact = file;
|
||||
plugin.signedBy = signedBy;
|
||||
if (replaced) {
|
||||
pluginLogger.info(
|
||||
`Plugin ${id} updated from ${replaced.manifest.version} to ${plugin.manifest.version}`,
|
||||
{ operation: "plugin_load" },
|
||||
);
|
||||
}
|
||||
return plugin;
|
||||
} catch (error) {
|
||||
if (replaced) this.plugins.set(id, replaced);
|
||||
throw error;
|
||||
}
|
||||
} finally {
|
||||
await fs.promises.rm(staging, { recursive: true, force: true });
|
||||
}
|
||||
@@ -427,9 +494,12 @@ export class PluginLoader {
|
||||
plugin.state = "activating";
|
||||
plugin.lastError = null;
|
||||
|
||||
const entry = pathToFileUrlString(
|
||||
getPluginBackendEntry(plugin.dir, plugin.manifest),
|
||||
);
|
||||
const entryPath = getPluginBackendEntry(plugin.dir, plugin.manifest);
|
||||
// Node caches ESM by URL, so a plugin replaced without a restart would
|
||||
// otherwise keep running the old code.
|
||||
const entry = `${pathToFileUrlString(entryPath)}?v=${encodeURIComponent(
|
||||
`${plugin.manifest.version}-${fileStamp(entryPath)}`,
|
||||
)}`;
|
||||
|
||||
try {
|
||||
const imported = (await import(entry)) as {
|
||||
@@ -639,27 +709,12 @@ export class PluginLoader {
|
||||
this.options.onFailed?.(plugin);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wraps a plugin callback so a throw is counted rather than escaping into
|
||||
* whichever core call site invoked it.
|
||||
*/
|
||||
wrap<Args extends unknown[], Result>(
|
||||
pluginId: string,
|
||||
fn: (...args: Args) => Result | Promise<Result>,
|
||||
): (...args: Args) => Promise<Result | undefined> {
|
||||
return async (...args: Args) => {
|
||||
try {
|
||||
return await fn(...args);
|
||||
} catch (error) {
|
||||
await this.reportError(pluginId, error);
|
||||
return undefined;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
/** Clears the error budget and starts the plugin again. */
|
||||
async retry(pluginId: string): Promise<void> {
|
||||
const plugin = this.requirePlugin(pluginId);
|
||||
if (plugin.state === "activating") {
|
||||
throw new Error(`Plugin ${pluginId} is still activating`);
|
||||
}
|
||||
plugin.errorTimestamps = [];
|
||||
plugin.lastError = null;
|
||||
plugin.state = "loaded";
|
||||
@@ -709,6 +764,43 @@ function withTimeout<T>(
|
||||
});
|
||||
}
|
||||
|
||||
const PLUGIN_ID_PATTERN = /^[a-z0-9][a-z0-9-]*$/;
|
||||
|
||||
/**
|
||||
* Throws unless an artifact may replace the bundled plugin with the same id:
|
||||
* a trusted key signed it and its version is strictly higher, so an old
|
||||
* signed build can never be replayed as a downgrade.
|
||||
*/
|
||||
export function assertBundledUpdate(
|
||||
id: string,
|
||||
version: unknown,
|
||||
signedBy: string | undefined,
|
||||
bundledVersion: string | undefined,
|
||||
): void {
|
||||
if (!signedBy) {
|
||||
throw new Error(
|
||||
`a bundled plugin already uses the id "${id}"; only a signed update can replace it`,
|
||||
);
|
||||
}
|
||||
if (typeof version !== "string" || !semver.valid(version)) {
|
||||
throw new Error(`update for "${id}" has no valid version`);
|
||||
}
|
||||
if (bundledVersion && !semver.gt(version, bundledVersion)) {
|
||||
throw new Error(
|
||||
`update for "${id}" is ${version}, which is not newer than the bundled ${bundledVersion}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
function fileStamp(filePath: string): string {
|
||||
try {
|
||||
const stat = fs.statSync(filePath);
|
||||
return `${stat.size}-${Math.round(stat.mtimeMs)}`;
|
||||
} catch {
|
||||
return "0";
|
||||
}
|
||||
}
|
||||
|
||||
function pathToFileUrlString(filePath: string): string {
|
||||
const resolved = path.resolve(filePath).replace(/\\/g, "/");
|
||||
return `file://${resolved.startsWith("/") ? "" : "/"}${resolved}`;
|
||||
|
||||
@@ -11,6 +11,9 @@ export {
|
||||
validateManifest,
|
||||
parseManifest,
|
||||
SUPPORTED_PLUGIN_API_VERSION,
|
||||
PLUGIN_API_VERSION,
|
||||
isApiCompatible,
|
||||
isTermixCompatible,
|
||||
SYSTEM_ROLE_NAMES,
|
||||
RESERVED_PERMISSION_PREFIXES,
|
||||
qualifyPermission,
|
||||
|
||||
@@ -20,6 +20,7 @@ import { pluginLogger } from "../utils/logger.js";
|
||||
import type { DatabaseDialect } from "../database/db/dialect.js";
|
||||
import { LEGACY_TABLE_OWNERS } from "@termix/plugin-sdk/db";
|
||||
import {
|
||||
collectOwnedIndexes,
|
||||
findUnownedTableWrites,
|
||||
splitStatements,
|
||||
} from "@termix/plugin-sdk/ddl";
|
||||
@@ -28,9 +29,6 @@ import {
|
||||
// migration exactly the way this runner does.
|
||||
export { LEGACY_TABLE_OWNERS, splitStatements };
|
||||
|
||||
/** Dialect directory names, matching DatabaseDialect exactly. */
|
||||
const DIALECTS: readonly DatabaseDialect[] = ["sqlite", "postgres", "mysql"];
|
||||
|
||||
const FILE_PATTERN = /^(\d{4})_([a-z0-9_]+)\.sql$/;
|
||||
|
||||
export interface PluginMigration {
|
||||
@@ -120,7 +118,7 @@ export function readMigrations(
|
||||
export function assertOwnedTables(
|
||||
pluginId: string,
|
||||
sql: string,
|
||||
options: { bundled?: boolean } = {},
|
||||
options: { bundled?: boolean; ownedIndexes?: ReadonlySet<string> } = {},
|
||||
): void {
|
||||
const legacy = new Set(
|
||||
options.bundled
|
||||
@@ -129,16 +127,19 @@ export function assertOwnedTables(
|
||||
.map(([table]) => table)
|
||||
: [],
|
||||
);
|
||||
const problems = findUnownedTableWrites(pluginId, sql, legacy).map(
|
||||
(problem) => {
|
||||
// Name the owner when the table is another plugin's legacy table.
|
||||
const table = /writes to "([^"]+)"/.exec(problem)?.[1];
|
||||
const owner = table ? LEGACY_TABLE_OWNERS[table] : undefined;
|
||||
return owner && owner !== pluginId
|
||||
? `may not touch "${table}": it belongs to the "${owner}" plugin`
|
||||
: problem;
|
||||
},
|
||||
);
|
||||
const problems = findUnownedTableWrites(
|
||||
pluginId,
|
||||
sql,
|
||||
legacy,
|
||||
options.ownedIndexes,
|
||||
).map((problem) => {
|
||||
// Name the owner when the table is another plugin's legacy table.
|
||||
const table = /writes to "([^"]+)"/.exec(problem)?.[1];
|
||||
const owner = table ? LEGACY_TABLE_OWNERS[table] : undefined;
|
||||
return owner && owner !== pluginId
|
||||
? `may not touch "${table}": it belongs to the "${owner}" plugin`
|
||||
: problem;
|
||||
});
|
||||
if (problems.length > 0) {
|
||||
throw new Error(
|
||||
`Plugin "${pluginId}" migration refused: ${problems.join("; ")}`,
|
||||
@@ -197,9 +198,13 @@ export async function applyPluginMigrations(
|
||||
|
||||
if (pending.length === 0) return [];
|
||||
|
||||
const ownedIndexes = collectOwnedIndexes(
|
||||
pluginId,
|
||||
migrations.map((migration) => migration.sql),
|
||||
);
|
||||
const run = async () => {
|
||||
for (const migration of pending) {
|
||||
assertOwnedTables(pluginId, migration.sql, options);
|
||||
assertOwnedTables(pluginId, migration.sql, { ...options, ownedIndexes });
|
||||
const statements = splitStatements(migration.sql);
|
||||
await runner.execute(statements);
|
||||
await runner.record(pluginId, migration);
|
||||
|
||||
@@ -1,19 +1,11 @@
|
||||
/**
|
||||
* A small main-thread service registry.
|
||||
* A small in-process key/value registry for live objects plugins hand each
|
||||
* other (a guest-link resolver, a host import normalizer). Services with
|
||||
* versions and permissions go through service-registry.ts instead.
|
||||
*
|
||||
* This exists so a module can depend on a capability without depending on
|
||||
* whichever module happens to provide it. Two uses today:
|
||||
*
|
||||
* - core registers the SSH connection pool as "ssh.transport" at boot, so
|
||||
* everything that needs a pooled connection resolves a provider rather
|
||||
* than importing the pool directly.
|
||||
* - the session-sharing plugin publishes "sessions.sharing.guests", guest
|
||||
* link resolution for the terminal, keyed by a token rather than a user.
|
||||
*
|
||||
* Deliberately NOT reachable from a worker plugin's ctx. Values here are live
|
||||
* objects -- pools, session managers, sockets -- and handing one across the
|
||||
* postMessage boundary is exactly what that boundary exists to prevent. Only
|
||||
* in-process first-party plugins get a reference to this.
|
||||
* ctx.registry only lets a plugin provide or revoke keys under its own id
|
||||
* ("<pluginId>.something"); anyone may consume. Core also reads a few
|
||||
* well-known per-plugin keys, such as "<pluginId>.hostImportNormalizer".
|
||||
*
|
||||
* `consume` returns undefined rather than throwing when nothing is registered.
|
||||
* A provider can disappear when its plugin is disabled, and callers are
|
||||
|
||||
@@ -23,6 +23,10 @@ export interface UserOwnedTable {
|
||||
/** The legacy core name the table adopted, for older export files. */
|
||||
legacyName?: string;
|
||||
userColumn: string;
|
||||
/** Other refUser columns, such as a grantee or the user who shared. */
|
||||
otherUserColumns: string[];
|
||||
/** refHost columns. */
|
||||
hostColumns: string[];
|
||||
/** Columns that travel, as SQL names. The primary key and secrets do not. */
|
||||
columns: string[];
|
||||
definition: PluginTableDefinition;
|
||||
@@ -39,6 +43,13 @@ export function listUserOwnedTables(): UserOwnedTable[] {
|
||||
table: prefixedTableName(pluginId, definition.name),
|
||||
legacyName: definition.adopts,
|
||||
userColumn: columnName(user[0], user[1]),
|
||||
otherUserColumns: entries
|
||||
.filter(([, column]) => column.type === "refUser")
|
||||
.slice(1)
|
||||
.map(([property, column]) => columnName(property, column)),
|
||||
hostColumns: entries
|
||||
.filter(([, column]) => column.type === "refHost")
|
||||
.map(([property, column]) => columnName(property, column)),
|
||||
columns: entries
|
||||
.filter(
|
||||
([, column]) =>
|
||||
@@ -130,10 +141,40 @@ export interface PluginRowsImport {
|
||||
errors: string[];
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether a row from an import file may be written as `userId`. The file is
|
||||
* user input, so a reference to anyone else's user or host is not trusted:
|
||||
* other user columns must point back at the row's own owner, and host columns
|
||||
* at a host the importing user owns.
|
||||
*/
|
||||
async function rowReferencesAllowed(
|
||||
owned: UserOwnedTable,
|
||||
source: Row,
|
||||
userId: string,
|
||||
): Promise<boolean> {
|
||||
const owner = source[owned.userColumn];
|
||||
for (const column of owned.otherUserColumns) {
|
||||
const value = source[column];
|
||||
if (value !== null && value !== undefined && value !== owner) return false;
|
||||
}
|
||||
for (const column of owned.hostColumns) {
|
||||
const value = source[column];
|
||||
if (value === null || value === undefined) continue;
|
||||
const rows = await selectRows(
|
||||
sql`SELECT 1 FROM ${sql.identifier("ssh_data")} WHERE ${sql.identifier(
|
||||
"id",
|
||||
)} = ${value} AND ${sql.identifier("user_id")} = ${userId} LIMIT 1`,
|
||||
);
|
||||
if (rows.length === 0) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Copies plugin rows from an import file into this server as `userId`. A row
|
||||
* matching one the user already has is skipped. Files from before 2.9 carry
|
||||
* the legacy table names, which are read too.
|
||||
* matching one the user already has is skipped, and so is one that points at
|
||||
* another user or at a host the importer does not own. Files from before 2.9
|
||||
* carry the legacy table names, which are read too.
|
||||
*/
|
||||
export async function importUserPluginRows(
|
||||
importDb: SqliteLike,
|
||||
@@ -157,10 +198,18 @@ export async function importUserPluginRows(
|
||||
const columns = owned.columns.filter(
|
||||
(column) => column === owned.userColumn || source[column] !== undefined,
|
||||
);
|
||||
const owner = source[owned.userColumn];
|
||||
const values = columns.map((column) =>
|
||||
column === owned.userColumn ? userId : source[column],
|
||||
column === owned.userColumn ||
|
||||
(owned.otherUserColumns.includes(column) && source[column] === owner)
|
||||
? userId
|
||||
: source[column],
|
||||
);
|
||||
try {
|
||||
if (!(await rowReferencesAllowed(owned, source, userId))) {
|
||||
summary.skipped++;
|
||||
continue;
|
||||
}
|
||||
const existing = await selectRows(
|
||||
sql`SELECT 1 FROM ${sql.identifier(owned.table)} WHERE ${sql.join(
|
||||
columns.map((column, index) =>
|
||||
|
||||
@@ -49,8 +49,13 @@ interface Route {
|
||||
declared: readonly string[];
|
||||
handler?: PluginWebSocketHandler;
|
||||
rawHandler?: RawUpgradeHandler;
|
||||
/** Sockets handed to rawHandler, destroyed when the route goes away. */
|
||||
rawSockets?: Set<Duplex>;
|
||||
}
|
||||
|
||||
/** Largest frame a plugin socket accepts. ws defaults to 100MB. */
|
||||
const MAX_PAYLOAD_BYTES = 64 * 1024 * 1024;
|
||||
|
||||
/** Keyed by "<pluginId>:<path>". */
|
||||
const routes = new Map<string, Route>();
|
||||
|
||||
@@ -177,6 +182,11 @@ async function handlePluginUpgrade(
|
||||
|
||||
if (route.rawHandler) {
|
||||
const rawHandler = route.rawHandler;
|
||||
const tracked = route.rawSockets;
|
||||
if (tracked) {
|
||||
tracked.add(socket);
|
||||
socket.once("close", () => tracked.delete(socket));
|
||||
}
|
||||
asCaller(() => rawHandler(request, socket, head, userId));
|
||||
return true;
|
||||
}
|
||||
@@ -270,7 +280,10 @@ export function registerPluginWsRoute(
|
||||
handler,
|
||||
});
|
||||
if (options.public) auditPublicSocket(pluginId, normalized);
|
||||
servers.set(routeKey, new WebSocketServer({ noServer: true }));
|
||||
servers.set(
|
||||
routeKey,
|
||||
new WebSocketServer({ noServer: true, maxPayload: MAX_PAYLOAD_BYTES }),
|
||||
);
|
||||
|
||||
pluginLogger.info(`Mounted /plugin-ws/${pluginId}${normalized}`, {
|
||||
operation: "plugin_ws_mount",
|
||||
@@ -301,6 +314,7 @@ export function registerPluginWsUpgrade(
|
||||
options,
|
||||
declared,
|
||||
rawHandler: handler,
|
||||
rawSockets: new Set(),
|
||||
});
|
||||
if (options.public) auditPublicSocket(pluginId, normalized);
|
||||
|
||||
@@ -312,8 +326,18 @@ export function registerPluginWsUpgrade(
|
||||
}
|
||||
|
||||
function disposeRoute(routeKey: string, pluginId: string, path: string): void {
|
||||
const route = routes.get(routeKey);
|
||||
routes.delete(routeKey);
|
||||
|
||||
for (const socket of route?.rawSockets ?? []) {
|
||||
try {
|
||||
socket.destroy();
|
||||
} catch {
|
||||
// Already gone.
|
||||
}
|
||||
}
|
||||
route?.rawSockets?.clear();
|
||||
|
||||
const server = servers.get(routeKey);
|
||||
if (server) {
|
||||
servers.delete(routeKey);
|
||||
|
||||
Reference in new issue
Block a user