From 181574d66f0dcaeb5c49326e68322ec0429025de Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 18 May 2026 14:04:18 -0500 Subject: [PATCH] Fix plugin bridge invocation scope propagation --- packages/plugins/sdk/README.md | 2 +- .../plugins/sdk/src/host-client-factory.ts | 105 +++++++++++- packages/plugins/sdk/src/index.ts | 4 + packages/plugins/sdk/src/protocol.ts | 49 ++++++ packages/plugins/sdk/src/worker-rpc-host.ts | 51 ++++-- .../sdk/tests/host-client-factory.test.ts | 79 +++++++++ .../plugins/sdk/tests/worker-rpc-host.test.ts | 159 +++++++++++++++++- .../plugin-worker-invocation-scope.cjs | 100 +++++++++++ .../__tests__/plugin-worker-manager.test.ts | 89 ++++++++++ server/src/routes/plugins.ts | 12 +- server/src/services/plugin-worker-manager.ts | 122 +++++++++++++- 11 files changed, 740 insertions(+), 32 deletions(-) create mode 100644 packages/plugins/sdk/tests/host-client-factory.test.ts create mode 100644 server/src/__tests__/fixtures/plugin-worker-invocation-scope.cjs diff --git a/packages/plugins/sdk/README.md b/packages/plugins/sdk/README.md index f94198af9df..b9a02d5c149 100644 --- a/packages/plugins/sdk/README.md +++ b/packages/plugins/sdk/README.md @@ -134,7 +134,7 @@ Subscribe in `setup` with `ctx.events.on(name, handler)` or `ctx.events.on(name, **Filter (optional):** Pass a second argument to `on()`: `{ projectId?, companyId?, agentId? }` so the host only delivers matching events. -**Company context:** Events still carry `companyId` for company-scoped data, but plugin installation and activation are instance-wide in the current runtime. +**Company context:** Events still carry `companyId` for company-scoped data, but plugin installation and activation are instance-wide in the current runtime. Access and authorization host services require an active company-scoped invocation such as an event, API route, tool run, environment call, or UI bridge call; the requested `companyId` must match that active scope. ## Scheduled (recurring) jobs diff --git a/packages/plugins/sdk/src/host-client-factory.ts b/packages/plugins/sdk/src/host-client-factory.ts index 9482c24fa78..3264afbccb9 100644 --- a/packages/plugins/sdk/src/host-client-factory.ts +++ b/packages/plugins/sdk/src/host-client-factory.ts @@ -49,7 +49,7 @@ */ import type { PluginCapability } from "@paperclipai/shared"; -import type { WorkerToHostMethods, WorkerToHostMethodName } from "./protocol.js"; +import type { WorkerHostCallContext, WorkerToHostMethods, WorkerToHostMethodName } from "./protocol.js"; import { PLUGIN_RPC_ERROR_CODES } from "./protocol.js"; // --------------------------------------------------------------------------- @@ -73,6 +73,19 @@ export class CapabilityDeniedError extends Error { } } +/** + * Thrown when a worker→host call asks for company-scoped data outside the + * company authorized for the current top-level plugin invocation. + */ +export class InvocationScopeDeniedError extends Error { + override readonly name = "InvocationScopeDeniedError"; + readonly code = PLUGIN_RPC_ERROR_CODES.CAPABILITY_DENIED; + + constructor(pluginId: string, method: string, message: string) { + super(`Plugin "${pluginId}" is not allowed to perform "${method}": ${message}`); + } +} + // --------------------------------------------------------------------------- // Host service interfaces // --------------------------------------------------------------------------- @@ -314,6 +327,7 @@ export interface HostClientFactoryOptions { */ type HostHandler = ( params: WorkerToHostMethods[M][0], + context?: WorkerHostCallContext, ) => Promise; /** @@ -501,6 +515,81 @@ export function createHostClientHandlers( const { pluginId, services } = options; const capabilitySet = new Set(options.capabilities); + type CompanyScopeRequest = + | { kind: "none" } + | { kind: "single"; companyId: string } + | { kind: "all" }; + + const noCompanyScope: CompanyScopeRequest = { kind: "none" }; + + function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); + } + + function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; + } + + function requestedCompanyScope( + method: WorkerToHostMethodName, + params: unknown, + ): CompanyScopeRequest { + if (method === "companies.list") return { kind: "all" }; + if (!isRecord(params)) return noCompanyScope; + + const companyId = readNonEmptyString(params.companyId); + if (companyId) return { kind: "single", companyId }; + + if (params.scopeKind === "company") { + const scopeId = readNonEmptyString(params.scopeId); + return scopeId ? { kind: "single", companyId: scopeId } : { kind: "all" }; + } + + if (method === "events.subscribe" && isRecord(params.filter)) { + const filterCompanyId = readNonEmptyString(params.filter.companyId); + if (filterCompanyId) return { kind: "single", companyId: filterCompanyId }; + } + + return noCompanyScope; + } + + function requireInvocationCompanyScope( + method: WorkerToHostMethodName, + params: unknown, + context?: WorkerHostCallContext, + ): void { + const requested = requestedCompanyScope(method, params); + if (requested.kind === "none") return; + + if (context?.invalidInvocationScope) { + throw new InvocationScopeDeniedError( + pluginId, + method, + "the worker referenced a missing, expired, or unknown invocation scope", + ); + } + + const allowedCompanyId = readNonEmptyString(context?.invocationScope?.companyId); + if (!allowedCompanyId) return; + + if (requested.kind === "all") { + if (method === "companies.list") return; + throw new InvocationScopeDeniedError( + pluginId, + method, + `the current invocation is scoped to company "${allowedCompanyId}"`, + ); + } + + if (requested.companyId !== allowedCompanyId) { + throw new InvocationScopeDeniedError( + pluginId, + method, + `requested company "${requested.companyId}" but the current invocation is scoped to company "${allowedCompanyId}"`, + ); + } + } + /** * Assert that the plugin has the required capability for a method. * Throws `CapabilityDeniedError` if the capability is missing. @@ -525,9 +614,10 @@ export function createHostClientHandlers( method: M, handler: HostHandler, ): HostHandler { - return async (params: WorkerToHostMethods[M][0]) => { + return async (params: WorkerToHostMethods[M][0], context?: WorkerHostCallContext) => { requireCapability(method); - return handler(params); + requireInvocationCompanyScope(method, params, context); + return handler(params, context); }; } @@ -631,8 +721,13 @@ export function createHostClientHandlers( }), // Companies - "companies.list": gated("companies.list", async (params) => { - return services.companies.list(params); + "companies.list": gated("companies.list", async (params, context) => { + const rows = await services.companies.list(params); + const allowedCompanyId = readNonEmptyString(context?.invocationScope?.companyId); + if (!allowedCompanyId) return rows; + return rows.filter((company) => + isRecord(company) && company.id === allowedCompanyId, + ) as WorkerToHostMethods["companies.list"][1]; }), "companies.get": gated("companies.get", async (params) => { return services.companies.get(params); diff --git a/packages/plugins/sdk/src/index.ts b/packages/plugins/sdk/src/index.ts index 27664664b58..2ca0fa9a7c3 100644 --- a/packages/plugins/sdk/src/index.ts +++ b/packages/plugins/sdk/src/index.ts @@ -58,6 +58,7 @@ export { createHostClientHandlers, getRequiredCapability, CapabilityDeniedError, + InvocationScopeDeniedError, } from "./host-client-factory.js"; // JSON-RPC protocol helpers and constants @@ -137,6 +138,9 @@ export type { JsonRpcMessage, JsonRpcErrorCode, PluginRpcErrorCode, + PluginInvocationScope, + PluginInvocationContext, + WorkerHostCallContext, InitializeParams, InitializeResult, ConfigChangedParams, diff --git a/packages/plugins/sdk/src/protocol.ts b/packages/plugins/sdk/src/protocol.ts index 99c47a00376..48075b2f6b9 100644 --- a/packages/plugins/sdk/src/protocol.ts +++ b/packages/plugins/sdk/src/protocol.ts @@ -104,6 +104,14 @@ export interface JsonRpcRequest< readonly method: TMethod; /** Structured parameters for the method call. */ readonly params: TParams; + /** + * Host-issued metadata for the top-level plugin invocation that is currently + * executing. The worker treats this as opaque and echoes only the id on + * worker→host calls made from the same async execution context. + */ + readonly paperclipInvocation?: PluginInvocationContext; + /** Opaque top-level invocation id echoed by worker→host requests. */ + readonly paperclipInvocationId?: string; } /** @@ -164,6 +172,13 @@ export interface JsonRpcNotification< readonly method: TMethod; /** Structured parameters for the notification. */ readonly params: TParams; + /** + * Host-issued metadata for host→worker push notifications such as events. + * Worker→host notifications echo only `paperclipInvocationId`. + */ + readonly paperclipInvocation?: PluginInvocationContext; + /** Opaque top-level invocation id echoed by worker→host notifications. */ + readonly paperclipInvocationId?: string; } /** @@ -225,6 +240,36 @@ export const PLUGIN_RPC_ERROR_CODES = { export type PluginRpcErrorCode = (typeof PLUGIN_RPC_ERROR_CODES)[keyof typeof PLUGIN_RPC_ERROR_CODES]; +// --------------------------------------------------------------------------- +// Invocation scope metadata +// --------------------------------------------------------------------------- + +/** + * Company scope attached by the host to one top-level plugin invocation. + * Absence of this metadata means the invocation is instance/global scoped. + */ +export interface PluginInvocationScope { + companyId: string; +} + +/** + * Opaque invocation metadata generated by the host. Workers must not derive or + * mutate this. They only echo the id on nested worker→host RPC calls. + */ +export interface PluginInvocationContext { + id: string; + scope: PluginInvocationScope; +} + +/** + * Context provided to host-side worker→host handlers after the worker echoes a + * host-issued invocation id. + */ +export interface WorkerHostCallContext { + invocationScope?: PluginInvocationScope | null; + invalidInvocationScope?: boolean; +} + // --------------------------------------------------------------------------- // Host → Worker Method Signatures (§13 Host-Worker Protocol) // --------------------------------------------------------------------------- @@ -310,6 +355,8 @@ export interface RunJobParams { export interface GetDataParams { /** Plugin-defined data key (e.g. `"sync-health"`). */ key: string; + /** Host-authorized active company scope, when this bridge call is company-scoped. */ + companyId?: string | null; /** Context and query parameters from the UI. */ params: Record; /** Optional launcher/container metadata from the host render environment. */ @@ -324,6 +371,8 @@ export interface GetDataParams { export interface PerformActionParams { /** Plugin-defined action key (e.g. `"resync"`). */ key: string; + /** Host-authorized active company scope, when this bridge call is company-scoped. */ + companyId?: string | null; /** Action parameters from the UI. */ params: Record; /** Optional launcher/container metadata from the host render environment. */ diff --git a/packages/plugins/sdk/src/worker-rpc-host.ts b/packages/plugins/sdk/src/worker-rpc-host.ts index effad802013..a50b6d8f14c 100644 --- a/packages/plugins/sdk/src/worker-rpc-host.ts +++ b/packages/plugins/sdk/src/worker-rpc-host.ts @@ -35,6 +35,7 @@ */ import fs from "node:fs"; +import { AsyncLocalStorage } from "node:async_hooks"; import path from "node:path"; import { createInterface, type Interface as ReadlineInterface } from "node:readline"; import { fileURLToPath } from "node:url"; @@ -66,6 +67,7 @@ import type { } from "./types.js"; import type { JsonRpcId, + JsonRpcNotification, JsonRpcRequest, JsonRpcResponse, InitializeParams, @@ -85,6 +87,7 @@ import type { PluginEnvironmentResumeLeaseParams, PluginEnvironmentValidateConfigParams, PluginEnvironmentProbeParams, + PluginInvocationContext, WorkerToHostMethodName, WorkerToHostMethods, } from "./protocol.js"; @@ -279,6 +282,7 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost let manifest: PaperclipPluginManifestV1 | null = null; let currentConfig: Record = {}; let databaseNamespace: string | null = null; + const invocationContextStorage = new AsyncLocalStorage(); // Plugin handler registrations (populated during setup()) const eventHandlers: EventRegistration[] = []; @@ -365,7 +369,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }); try { - const request = createRequest(method, params, id); + const activeInvocation = invocationContextStorage.getStore(); + const request = { + ...createRequest(method, params, id), + ...(activeInvocation ? { paperclipInvocationId: activeInvocation.id } : {}), + }; sendMessage(request); } catch (err) { settle(reject, err instanceof Error ? err : new Error(String(err))); @@ -378,7 +386,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost */ function notifyHost(method: string, params: unknown): void { try { - sendMessage(createNotification(method, params)); + const activeInvocation = invocationContextStorage.getStore(); + sendMessage({ + ...createNotification(method, params), + ...(activeInvocation ? { paperclipInvocationId: activeInvocation.id } : {}), + }); } catch { // Swallow — the host may have closed stdin } @@ -1254,7 +1266,10 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost const { id, method, params } = request; try { - const result = await dispatchMethod(method, params); + const invoke = () => dispatchMethod(method, params); + const result = request.paperclipInvocation + ? await invocationContextStorage.run(request.paperclipInvocation, invoke) + : await invoke(); sendMessage(createSuccessResponse(id, result ?? null)); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); @@ -1492,11 +1507,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (!handler) { throw new Error(`No data handler registered for key "${params.key}"`); } - return handler( - params.renderEnvironment === undefined - ? params.params - : { ...params.params, renderEnvironment: params.renderEnvironment }, - ); + return handler({ + ...params.params, + ...(params.companyId === undefined ? {} : { companyId: params.companyId }), + ...(params.renderEnvironment === undefined ? {} : { renderEnvironment: params.renderEnvironment }), + }); } async function handlePerformAction(params: PerformActionParams): Promise { @@ -1504,11 +1519,11 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost if (!handler) { throw new Error(`No action handler registered for key "${params.key}"`); } - return handler( - params.renderEnvironment === undefined - ? params.params - : { ...params.params, renderEnvironment: params.renderEnvironment }, - ); + return handler({ + ...params.params, + ...(params.companyId === undefined ? {} : { companyId: params.companyId }), + ...(params.renderEnvironment === undefined ? {} : { renderEnvironment: params.renderEnvironment }), + }); } async function handleExecuteTool(params: ExecuteToolParams): Promise { @@ -1676,14 +1691,20 @@ export function startWorkerRpcHost(options: WorkerRpcHostOptions): WorkerRpcHost }); } else if (isJsonRpcNotification(message)) { // Dispatch host→worker push notifications - const notif = message as { method: string; params?: unknown }; + const notif = message as JsonRpcNotification & { method: string; params?: unknown }; + const runNotification = (fn: () => void | Promise) => { + if (notif.paperclipInvocation) { + return invocationContextStorage.run(notif.paperclipInvocation, fn); + } + return fn(); + }; if (notif.method === "agents.sessions.event" && notif.params) { const event = notif.params as AgentSessionEvent; const cb = sessionEventCallbacks.get(event.sessionId); if (cb) cb(event); } else if (notif.method === "onEvent" && notif.params) { // Plugin event bus notifications — dispatch to registered event handlers - handleOnEvent(notif.params as OnEventParams).catch((err) => { + Promise.resolve(runNotification(() => handleOnEvent(notif.params as OnEventParams))).catch((err) => { notifyHost("log", { level: "error", message: `Failed to handle event notification: ${err instanceof Error ? err.message : String(err)}`, diff --git a/packages/plugins/sdk/tests/host-client-factory.test.ts b/packages/plugins/sdk/tests/host-client-factory.test.ts new file mode 100644 index 00000000000..1c7d3da0e70 --- /dev/null +++ b/packages/plugins/sdk/tests/host-client-factory.test.ts @@ -0,0 +1,79 @@ +import { describe, expect, it, vi } from "vitest"; + +import type { HostServices } from "../src/host-client-factory.js"; +import { + createHostClientHandlers, + InvocationScopeDeniedError, +} from "../src/host-client-factory.js"; + +describe("createHostClientHandlers invocation company scope", () => { + it("rejects company-scoped host calls outside the current invocation company", async () => { + const projectsList = vi.fn(async () => []); + const services = { + projects: { + list: projectsList, + }, + } as unknown as HostServices; + + const handlers = createHostClientHandlers({ + pluginId: "paperclip.test", + capabilities: ["projects.read"], + services, + }); + + await expect( + handlers["projects.list"]( + { companyId: "company-b" }, + { invocationScope: { companyId: "company-a" } }, + ), + ).rejects.toBeInstanceOf(InvocationScopeDeniedError); + expect(projectsList).not.toHaveBeenCalled(); + }); + + it("filters companies.list to the current invocation company", async () => { + const services = { + companies: { + list: vi.fn(async () => [ + { id: "company-a", name: "Company A" }, + { id: "company-b", name: "Company B" }, + ]), + }, + } as unknown as HostServices; + + const handlers = createHostClientHandlers({ + pluginId: "paperclip.test", + capabilities: ["companies.read"], + services, + }); + + await expect( + handlers["companies.list"]( + {}, + { invocationScope: { companyId: "company-a" } }, + ), + ).resolves.toEqual([{ id: "company-a", name: "Company A" }]); + }); + + it("rejects company-scope store access for a different company", async () => { + const stateGet = vi.fn(async () => null); + const services = { + state: { + get: stateGet, + }, + } as unknown as HostServices; + + const handlers = createHostClientHandlers({ + pluginId: "paperclip.test", + capabilities: ["plugin.state.read"], + services, + }); + + await expect( + handlers["state.get"]( + { scopeKind: "company", scopeId: "company-b", stateKey: "settings" }, + { invocationScope: { companyId: "company-a" } }, + ), + ).rejects.toBeInstanceOf(InvocationScopeDeniedError); + expect(stateGet).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/plugins/sdk/tests/worker-rpc-host.test.ts b/packages/plugins/sdk/tests/worker-rpc-host.test.ts index f8e0a38e5dc..b7f15781b5f 100644 --- a/packages/plugins/sdk/tests/worker-rpc-host.test.ts +++ b/packages/plugins/sdk/tests/worker-rpc-host.test.ts @@ -1,11 +1,26 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; +import { createInterface } from "node:readline"; +import { PassThrough } from "node:stream"; import { pathToFileURL } from "node:url"; import { afterEach, describe, expect, it } from "vitest"; -import { isWorkerEntrypoint } from "../src/worker-rpc-host.js"; +import { definePlugin } from "../src/define-plugin.js"; +import { + createRequest, + createErrorResponse, + createSuccessResponse, + isJsonRpcRequest, + isJsonRpcResponse, + parseMessage, + PLUGIN_RPC_ERROR_CODES, + serializeMessage, + type JsonRpcResponse, + type PluginInvocationContext, +} from "../src/protocol.js"; +import { isWorkerEntrypoint, startWorkerRpcHost } from "../src/worker-rpc-host.js"; describe("isWorkerEntrypoint", () => { const tempRoots: string[] = []; @@ -55,3 +70,145 @@ describe("isWorkerEntrypoint", () => { ).toBe(false); }); }); + +describe("worker invocation scope propagation", () => { + it("keeps overlapping company scopes local to each getData invocation", async () => { + const hostToWorker = new PassThrough(); + const workerToHost = new PassThrough(); + const hostReadline = createInterface({ input: workerToHost }); + const pending = new Map void>(); + const nestedInvocationIds: string[] = []; + const invocationCompanies = new Map([ + ["invocation-a", "company-a"], + ["invocation-b", "company-b"], + ]); + let releaseCompanyA: (() => void) | null = null; + let nextRequestId = 1; + + const plugin = definePlugin({ + async setup(ctx) { + ctx.data.register("probe", async (params) => { + if (params.label === "a") { + await new Promise((resolve) => { + releaseCompanyA = resolve; + }); + } + const company = await ctx.companies.get(String(params.requestedCompanyId)); + return { label: params.label, company }; + }); + }, + }); + + const worker = startWorkerRpcHost({ + plugin, + stdin: hostToWorker, + stdout: workerToHost, + }); + + function callWorker(method: string, params: unknown, invocation?: PluginInvocationContext) { + const id = `host-${nextRequestId++}`; + const request = { + ...createRequest(method, params, id), + ...(invocation ? { paperclipInvocation: invocation } : {}), + }; + const result = new Promise((resolve, reject) => { + pending.set(id, (response) => { + if ("error" in response && response.error) { + reject(new Error(response.error.message)); + return; + } + resolve((response as { result?: unknown }).result); + }); + }); + hostToWorker.write(serializeMessage(request)); + return result; + } + + hostReadline.on("line", (line) => { + const message = parseMessage(line); + if (isJsonRpcResponse(message)) { + pending.get(String(message.id))?.(message); + pending.delete(String(message.id)); + return; + } + + if (!isJsonRpcRequest(message)) return; + if (message.method !== "companies.get") return; + + const invocationId = (message as { paperclipInvocationId?: string }).paperclipInvocationId ?? ""; + const requestedCompanyId = (message.params as { companyId?: string }).companyId; + const allowedCompanyId = invocationCompanies.get(invocationId); + nestedInvocationIds.push(invocationId); + if (requestedCompanyId !== allowedCompanyId) { + hostToWorker.write(serializeMessage(createErrorResponse( + message.id, + PLUGIN_RPC_ERROR_CODES.CAPABILITY_DENIED, + `requested company "${requestedCompanyId}" but invocation "${invocationId}" is scoped to "${allowedCompanyId}"`, + ))); + return; + } + + hostToWorker.write(serializeMessage(createSuccessResponse(message.id, { + id: requestedCompanyId, + }))); + + if (invocationId === "invocation-b") { + releaseCompanyA?.(); + } + }); + + try { + await callWorker("initialize", { + manifest: { + id: "paperclip.scope-test", + apiVersion: 1, + version: "1.0.0", + displayName: "Scope test", + description: "Scope test", + author: "Paperclip", + categories: ["automation"], + capabilities: ["companies.read"], + entrypoints: { worker: "dist/worker.js" }, + }, + config: {}, + instanceInfo: { instanceId: "test", hostVersion: "0.0.0" }, + apiVersion: 1, + }); + + const companyARequest = callWorker( + "getData", + { + key: "probe", + companyId: "company-a", + params: { label: "a", requestedCompanyId: "company-b" }, + }, + { id: "invocation-a", scope: { companyId: "company-a" } }, + ); + const companyAExpectation = expect(companyARequest).rejects.toThrow( + /requested company "company-b"/, + ); + const companyBRequest = callWorker( + "getData", + { + key: "probe", + companyId: "company-b", + params: { label: "b", requestedCompanyId: "company-b" }, + }, + { id: "invocation-b", scope: { companyId: "company-b" } }, + ); + + await expect(companyBRequest).resolves.toEqual({ + label: "b", + company: { id: "company-b" }, + }); + await companyAExpectation; + + expect(nestedInvocationIds).toEqual(["invocation-b", "invocation-a"]); + } finally { + worker.stop(); + hostReadline.close(); + hostToWorker.destroy(); + workerToHost.destroy(); + } + }); +}); diff --git a/server/src/__tests__/fixtures/plugin-worker-invocation-scope.cjs b/server/src/__tests__/fixtures/plugin-worker-invocation-scope.cjs new file mode 100644 index 00000000000..6e54689fc3e --- /dev/null +++ b/server/src/__tests__/fixtures/plugin-worker-invocation-scope.cjs @@ -0,0 +1,100 @@ +const readline = require("node:readline"); + +let nextRequestId = 1; +const pendingNested = new Map(); + +function send(message) { + process.stdout.write(`${JSON.stringify(message)}\n`); +} + +function sendNestedHostRequest(originalRequest, invocationId) { + const nestedId = `nested-${nextRequestId++}`; + const params = originalRequest.params?.params ?? {}; + const mode = params.mode; + const requestedCompanyId = params.requestedCompanyId; + const nestedRequest = { + jsonrpc: "2.0", + id: nestedId, + method: "companies.get", + params: { + companyId: requestedCompanyId, + }, + }; + + if (mode === "echo") { + nestedRequest.paperclipInvocationId = invocationId; + } else if (mode === "unknown") { + nestedRequest.paperclipInvocationId = "unknown-invocation"; + } + + pendingNested.set(nestedId, originalRequest.id); + send(nestedRequest); +} + +const rl = readline.createInterface({ + input: process.stdin, + crlfDelay: Infinity, +}); + +rl.on("line", (line) => { + if (!line.trim()) return; + const message = JSON.parse(line); + + if (message.id && pendingNested.has(message.id)) { + const originalId = pendingNested.get(message.id); + pendingNested.delete(message.id); + if (message.error) { + send({ + jsonrpc: "2.0", + id: originalId, + error: message.error, + }); + return; + } + + send({ + jsonrpc: "2.0", + id: originalId, + result: message.result, + }); + return; + } + + const method = message && typeof message.method === "string" ? message.method : null; + + if (method === "initialize") { + send({ + jsonrpc: "2.0", + id: message.id, + result: { + ok: true, + supportedMethods: ["getData"], + }, + }); + return; + } + + if (method === "getData") { + sendNestedHostRequest(message, message.paperclipInvocation?.id); + return; + } + + if (method === "shutdown") { + send({ + jsonrpc: "2.0", + id: message.id, + result: {}, + }); + setImmediate(() => process.exit(0)); + return; + } + + send({ + jsonrpc: "2.0", + id: message.id, + error: { + code: -32601, + message: `Unhandled method: ${method}`, + }, + }); +}); diff --git a/server/src/__tests__/plugin-worker-manager.test.ts b/server/src/__tests__/plugin-worker-manager.test.ts index 4f578fda072..626ed448cba 100644 --- a/server/src/__tests__/plugin-worker-manager.test.ts +++ b/server/src/__tests__/plugin-worker-manager.test.ts @@ -3,7 +3,10 @@ import { fileURLToPath } from "node:url"; import { describe, expect, it, vi } from "vitest"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; import { + createHostClientHandlers, JsonRpcCallError, + PLUGIN_RPC_ERROR_CODES, + type HostServices, type HostToWorkerMethods, } from "@paperclipai/plugin-sdk"; import { @@ -14,6 +17,10 @@ import { const FIXTURES_DIR = path.join(path.dirname(fileURLToPath(import.meta.url)), "fixtures"); const DELAYED_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-delayed.cjs"); +const INVOCATION_SCOPE_WORKER_ENTRYPOINT = path.join( + FIXTURES_DIR, + "plugin-worker-invocation-scope.cjs", +); const TERMINATED_WORKER_ENTRYPOINT = path.join(FIXTURES_DIR, "plugin-worker-terminated.cjs"); const TEST_MANIFEST: PaperclipPluginManifestV1 = { @@ -178,4 +185,86 @@ describe("plugin-worker-manager stderr failure context", () => { await handle.stop().catch(() => undefined); } }); + + it("passes echoed invocation scope to worker-to-host handlers", async () => { + const companiesGet = vi.fn(async () => ({ id: "company-1" })); + const handle = createPluginWorkerHandle("test.plugin", { + entrypointPath: INVOCATION_SCOPE_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { + instanceId: "instance-1", + hostVersion: "1.0.0", + }, + apiVersion: 1, + hostHandlers: { + "companies.get": companiesGet, + }, + }); + + try { + await handle.start(); + + await expect(handle.call("getData", { + key: "probe", + companyId: "company-1", + params: { + mode: "echo", + requestedCompanyId: "company-1", + }, + } as HostToWorkerMethods["getData"][0])).resolves.toEqual({ id: "company-1" }); + + expect(companiesGet).toHaveBeenCalledWith( + { companyId: "company-1" }, + { invocationScope: { companyId: "company-1" } }, + ); + } finally { + await handle.stop().catch(() => undefined); + } + }); + + it("rejects missing or unknown invocation ids while a company invocation is active", async () => { + const companiesGet = vi.fn(async () => ({ id: "company-2" })); + const hostHandlers = createHostClientHandlers({ + pluginId: "test.plugin", + capabilities: ["companies.read"], + services: { + companies: { + get: companiesGet, + }, + } as unknown as HostServices, + }); + const handle = createPluginWorkerHandle("test.plugin", { + entrypointPath: INVOCATION_SCOPE_WORKER_ENTRYPOINT, + manifest: TEST_MANIFEST, + config: {}, + instanceInfo: { + instanceId: "instance-1", + hostVersion: "1.0.0", + }, + apiVersion: 1, + hostHandlers, + }); + + try { + await handle.start(); + + for (const mode of ["omit", "unknown"]) { + await expect(handle.call("getData", { + key: "probe", + companyId: "company-1", + params: { + mode, + requestedCompanyId: "company-2", + }, + } as HostToWorkerMethods["getData"][0])).rejects.toMatchObject({ + code: PLUGIN_RPC_ERROR_CODES.CAPABILITY_DENIED, + }); + } + + expect(companiesGet).not.toHaveBeenCalled(); + } finally { + await handle.stop().catch(() => undefined); + } + }); }); diff --git a/server/src/routes/plugins.ts b/server/src/routes/plugins.ts index 6abb7c56ec5..741fa1513fe 100644 --- a/server/src/routes/plugins.ts +++ b/server/src/routes/plugins.ts @@ -1116,7 +1116,7 @@ export function pluginRoutes( return; } - assertPluginBridgeScope(req, body.companyId); + const companyId = assertPluginBridgeScope(req, body.companyId); try { const result = await bridgeDeps.workerManager.call( @@ -1124,6 +1124,7 @@ export function pluginRoutes( "getData", { key: body.key, + ...(companyId ? { companyId } : {}), params: body.params ?? {}, renderEnvironment: body.renderEnvironment ?? null, }, @@ -1208,7 +1209,7 @@ export function pluginRoutes( return; } - assertPluginBridgeScope(req, body.companyId); + const companyId = assertPluginBridgeScope(req, body.companyId); try { const result = await bridgeDeps.workerManager.call( @@ -1216,6 +1217,7 @@ export function pluginRoutes( "performAction", { key: body.key, + ...(companyId ? { companyId } : {}), params: body.params ?? {}, renderEnvironment: body.renderEnvironment ?? null, }, @@ -1301,7 +1303,7 @@ export function pluginRoutes( renderEnvironment?: PluginLauncherRenderContextSnapshot | null; } | undefined; - assertPluginBridgeScope(req, body?.companyId); + const companyId = assertPluginBridgeScope(req, body?.companyId); try { const result = await bridgeDeps.workerManager.call( @@ -1309,6 +1311,7 @@ export function pluginRoutes( "getData", { key, + ...(companyId ? { companyId } : {}), params: body?.params ?? {}, renderEnvironment: body?.renderEnvironment ?? null, }, @@ -1390,7 +1393,7 @@ export function pluginRoutes( renderEnvironment?: PluginLauncherRenderContextSnapshot | null; } | undefined; - assertPluginBridgeScope(req, body?.companyId); + const companyId = assertPluginBridgeScope(req, body?.companyId); try { const result = await bridgeDeps.workerManager.call( @@ -1398,6 +1401,7 @@ export function pluginRoutes( "performAction", { key, + ...(companyId ? { companyId } : {}), params: body?.params ?? {}, renderEnvironment: body?.renderEnvironment ?? null, }, diff --git a/server/src/services/plugin-worker-manager.ts b/server/src/services/plugin-worker-manager.ts index 6e1d7689be2..81f21a91f85 100644 --- a/server/src/services/plugin-worker-manager.ts +++ b/server/src/services/plugin-worker-manager.ts @@ -19,6 +19,7 @@ */ import { fork, type ChildProcess } from "node:child_process"; +import { randomUUID } from "node:crypto"; import { EventEmitter } from "node:events"; import { createInterface, type Interface as ReadlineInterface } from "node:readline"; import type { PaperclipPluginManifestV1 } from "@paperclipai/shared"; @@ -39,9 +40,12 @@ import { } from "@paperclipai/plugin-sdk"; import type { JsonRpcId, + PluginInvocationContext, + PluginInvocationScope, JsonRpcResponse, JsonRpcRequest, JsonRpcNotification, + WorkerHostCallContext, HostToWorkerMethodName, HostToWorkerMethods, WorkerToHostMethodName, @@ -108,6 +112,7 @@ export type WorkerStatus = */ export type WorkerToHostHandler = ( params: WorkerToHostMethods[M][0], + context?: WorkerHostCallContext, ) => Promise; /** @@ -201,6 +206,11 @@ interface PendingRequest { sentAt: number; } +interface ActiveInvocation { + scope: PluginInvocationScope; + timer?: ReturnType; +} + // --------------------------------------------------------------------------- // PluginWorkerHandle — manages a single worker process // --------------------------------------------------------------------------- @@ -379,6 +389,7 @@ export function createPluginWorkerHandle( // Pending RPC requests awaiting a response const pendingRequests = new Map(); let nextRequestId = 1; + const activeInvocations = new Map(); // Optional methods reported by the worker during initialization let supportedMethods: string[] = []; @@ -475,13 +486,78 @@ export function createPluginWorkerHandle( pending.resolve(response); } + function readNonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; + } + + function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); + } + + function deriveInvocationScope( + method: HostToWorkerMethodName | string, + params: unknown, + ): PluginInvocationScope | null { + if (!isRecord(params)) return null; + + const directCompanyId = readNonEmptyString(params.companyId); + if (directCompanyId) return { companyId: directCompanyId }; + + if (method === "executeTool" && isRecord(params.runContext)) { + const companyId = readNonEmptyString(params.runContext.companyId); + return companyId ? { companyId } : null; + } + + if (method === "onEvent" && isRecord(params.event)) { + const companyId = readNonEmptyString(params.event.companyId); + return companyId ? { companyId } : null; + } + + return null; + } + + function registerInvocation(scope: PluginInvocationScope, ttlMs?: number): PluginInvocationContext { + const invocation: PluginInvocationContext = { + id: randomUUID(), + scope, + }; + const entry: ActiveInvocation = { scope }; + if (ttlMs !== undefined) { + entry.timer = setTimeout(() => { + activeInvocations.delete(invocation.id); + }, ttlMs); + if (entry.timer.unref) entry.timer.unref(); + } + activeInvocations.set(invocation.id, entry); + return invocation; + } + + function clearInvocation(invocation: PluginInvocationContext | null): void { + if (!invocation) return; + const entry = activeInvocations.get(invocation.id); + if (entry?.timer) clearTimeout(entry.timer); + activeInvocations.delete(invocation.id); + } + + function contextForWorkerMessage(message: JsonRpcRequest | JsonRpcNotification): WorkerHostCallContext { + const invocationId = readNonEmptyString( + (message as { paperclipInvocationId?: unknown }).paperclipInvocationId, + ); + if (!invocationId) { + return activeInvocations.size > 0 ? { invalidInvocationScope: true } : {}; + } + const entry = activeInvocations.get(invocationId); + if (!entry) return { invalidInvocationScope: true }; + return { invocationScope: entry.scope }; + } + /** * Handle a JSON-RPC request from the worker (worker→host call). */ async function handleWorkerRequest(request: JsonRpcRequest): Promise { const method = request.method as WorkerToHostMethodName; const handler = options.hostHandlers[method] as - | ((params: unknown) => Promise) + | ((params: unknown, context?: WorkerHostCallContext) => Promise) | undefined; if (!handler) { @@ -501,7 +577,7 @@ export function createPluginWorkerHandle( } try { - const result = await handler(request.params); + const result = await handler(request.params, contextForWorkerMessage(request)); sendMessage({ jsonrpc: JSONRPC_VERSION, id: request.id, @@ -509,12 +585,15 @@ export function createPluginWorkerHandle( }); } catch (err) { const errorMessage = err instanceof Error ? err.message : String(err); + const errorCode = typeof (err as { code?: unknown }).code === "number" + ? (err as { code: number }).code + : JSONRPC_ERROR_CODES.INTERNAL_ERROR; log.error({ method, err: errorMessage }, "host handler error"); try { sendMessage( createErrorResponse( request.id, - JSONRPC_ERROR_CODES.INTERNAL_ERROR, + errorCode, errorMessage, ), ); @@ -572,12 +651,28 @@ export function createPluginWorkerHandle( notification.method === "streams.close" ) { const params = (notification.params ?? {}) as Record; + const companyId = String(params.companyId ?? ""); + const context = contextForWorkerMessage(notification); + if (context.invalidInvocationScope) { + log.warn( + { method: notification.method, companyId }, + "dropping plugin stream notification with invalid invocation scope", + ); + return; + } + const allowedCompanyId = context.invocationScope?.companyId; + if (allowedCompanyId && companyId !== allowedCompanyId) { + log.warn( + { method: notification.method, companyId, allowedCompanyId }, + "dropping plugin stream notification outside invocation company scope", + ); + return; + } // Track open channels so we can emit synthetic close on crash if (notification.method === "streams.open") { const ch = String(params.channel ?? ""); - const co = String(params.companyId ?? ""); - if (ch) openStreamChannels.set(ch, co); + if (ch) openStreamChannels.set(ch, companyId); } else if (notification.method === "streams.close") { openStreamChannels.delete(String(params.channel ?? "")); } @@ -760,6 +855,10 @@ export function createPluginWorkerHandle( ); } pendingRequests.clear(); + for (const invocation of activeInvocations.values()) { + if (invocation.timer) clearTimeout(invocation.timer); + } + activeInvocations.clear(); } // ----------------------------------------------------------------------- @@ -1020,6 +1119,8 @@ export function createPluginWorkerHandle( const id = nextRequestId++; const timeout = Math.min(timeoutMs ?? rpcTimeoutMs, MAX_RPC_TIMEOUT_MS); + const invocationScope = deriveInvocationScope(method, params); + const invocation = invocationScope ? registerInvocation(invocationScope) : null; // Guard against double-settlement. When a process exits all pending // requests are rejected via rejectAllPending(), but the timeout timer @@ -1032,6 +1133,7 @@ export function createPluginWorkerHandle( settled = true; clearTimeout(timer); pendingRequests.delete(id); + clearInvocation(invocation); fn(value); }; @@ -1064,11 +1166,15 @@ export function createPluginWorkerHandle( pendingRequests.set(id, pending); try { - const request = createRequest(method, params, id); + const request = { + ...createRequest(method, params, id), + ...(invocation ? { paperclipInvocation: invocation } : {}), + }; sendMessage(request); } catch (err) { clearTimeout(timer); pendingRequests.delete(id); + clearInvocation(invocation); reject( new Error( `Failed to send "${method}" to worker: ${ @@ -1135,13 +1241,17 @@ export function createPluginWorkerHandle( notify(method: string, params: unknown) { if (status !== "running") return; + const invocationScope = deriveInvocationScope(method, params); + const invocation = invocationScope ? registerInvocation(invocationScope, MAX_RPC_TIMEOUT_MS) : null; try { sendMessage({ jsonrpc: JSONRPC_VERSION, method, params, + ...(invocation ? { paperclipInvocation: invocation } : {}), }); } catch { + clearInvocation(invocation); log.warn({ method }, "failed to send notification to worker"); } },