From 509ca80e4ecf2f33cbf6e0d43e38f7f70e4806ff Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 2 Nov 2026 06:03:32 +0100 Subject: [PATCH] fix(plugins): withhold binary plugin results after plugin access is revoked A binary RPC output or event skipped the outbound permission check its JSON form passes, so an RPC in flight when daemon.manage was revoked still delivered its bytes. Each frame now checks the permission of the JSON message it stands in for. --- .../src/server/plugins/runtime.posix.test.ts | 4 +- packages/server/src/server/session.test.ts | 37 +++++++++++++++++++ packages/server/src/server/session.ts | 26 ++++++++----- 3 files changed, 57 insertions(+), 10 deletions(-) diff --git a/packages/server/src/server/plugins/runtime.posix.test.ts b/packages/server/src/server/plugins/runtime.posix.test.ts index 01a7cf5217..bed5e98a00 100644 --- a/packages/server/src/server/plugins/runtime.posix.test.ts +++ b/packages/server/src/server/plugins/runtime.posix.test.ts @@ -1424,7 +1424,9 @@ export default function contribute(server: { registerProvider(provider: Provider const subscription = runtime.startSubscription("slow-subscribe", "feed", {}); const rejection = subscription.ready.catch((error: Error) => error); await vi.advanceTimersByTimeAsync(30_000); - await expect(rejection).resolves.toMatchObject({ message: expect.stringContaining("timed out") }); + await expect(rejection).resolves.toMatchObject({ + message: expect.stringContaining("timed out"), + }); const subscribe = sent.find((message) => message.type === "subscribe"); expect(sent).toContainEqual({ type: "unsubscribe", diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index acd292e895..4f02efad2e 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -348,6 +348,7 @@ interface SessionForTestOptions { messages?: unknown[]; targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>; binaryMessages?: Uint8Array[]; + clientCapabilities?: SessionOptions["clientCapabilities"]; pluginRuntime?: SessionOptions["pluginRuntime"]; orchestrationSkills?: SessionOptions["orchestrationSkills"]; workspaceLabelService?: WorkspaceLabelService; @@ -402,6 +403,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { messages.push(message), }), onBinaryMessage: createBinaryMessageHandler(options.binaryMessages), + clientCapabilities: options.clientCapabilities, logger, downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(), pushNotifications: options.pushNotifications ?? asPushNotifications(), @@ -624,6 +626,41 @@ test("losing daemon.manage releases the session's plugin subscriptions", async ( await vi.waitFor(() => expect(release).toHaveBeenCalledOnce()); }); +test("losing daemon.manage during a plugin RPC withholds its JSON and binary results", async () => { + const messages: SessionOutboundMessage[] = []; + const binaryMessages: Uint8Array[] = []; + const outputs = new Map void>(); + const pluginRuntime = { + subscribe: () => () => {}, + invokePluginRpc: (_pluginId: string, method: string) => + new Promise((resolve) => outputs.set(method, resolve)), + } as unknown as NonNullable; + const session = createSessionForTest({ + messages, + binaryMessages, + pluginRuntime, + clientCapabilities: { [CLIENT_CAPS.pluginBinaryPayloads]: true }, + }); + const invoke = (method: string) => + session.handleMessage({ + type: "plugin.rpc.invoke.request", + requestId: method, + pluginId: "example", + method, + input: {}, + }); + + const pending = [invoke("capture.json"), invoke("capture.bytes")]; + await vi.waitFor(() => expect(outputs.size).toBe(2)); + session.setPermissions(["daemon.read"]); + outputs.get("capture.json")?.({ width: 1 }); + outputs.get("capture.bytes")?.(new Uint8Array([1, 2, 3])); + await Promise.all(pending); + + expect(messages.filter((message) => message.type === "plugin.rpc.invoke.response")).toEqual([]); + expect(binaryMessages).toEqual([]); +}); + test("leaves out the client bundles a catalog request already holds", async () => { const messages: SessionOutboundMessage[] = []; const pluginRuntime: NonNullable = { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index c6cae8b74c..81c22082c7 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -2309,7 +2309,10 @@ export class Session { void this.delivery .releaseFamily("plugin-rpc") .catch((error) => - this.sessionLogger.error({ err: error }, "Failed to release revoked plugin subscriptions"), + this.sessionLogger.error( + { err: error }, + "Failed to release revoked plugin subscriptions", + ), ); } } @@ -2661,13 +2664,16 @@ export class Session { request: Extract, output: unknown, ): Promise { + const response: SessionOutboundMessage = { + type: "plugin.rpc.invoke.response", + payload: { requestId: request.requestId, output }, + }; if (!(output instanceof Uint8Array)) { - this.emit({ - type: "plugin.rpc.invoke.response", - payload: { requestId: request.requestId, output }, - }); + this.emit(response); return; } + // The frame is the binary form of the response, so it needs the response's permission. + if (!this.authorization.allowsOutbound(response)) return; if (!this.supports(CLIENT_CAPS.pluginBinaryPayloads) || this.onBinaryMessage === null) throw new Error( `Update Paseo to receive the binary output of plugin RPC ${request.pluginId}.${request.method}`, @@ -2711,13 +2717,15 @@ export class Session { }; subscription.attach({ event: (event) => { + const message: SessionOutboundMessage = { + type: "plugin.rpc.event", + payload: { subscriptionId: owner.responseId, event }, + }; if (!(event instanceof Uint8Array)) { - owner.emit({ - type: "plugin.rpc.event", - payload: { subscriptionId: owner.responseId, event }, - }); + owner.emit(message); return; } + if (!this.authorization.allowsOutbound(message)) return; if (this.onBinaryMessage === null) { subscription.release(); close("This connection cannot carry binary plugin events");