From d49434d984b0f0bf2ff6e9fd8dcdaa115c2a8ab8 Mon Sep 17 00:00:00 2001 From: FRFlo <62435410+FRFlo@users.noreply.github.com> Date: Sun, 18 Oct 2026 13:12:33 +0200 Subject: [PATCH 1/2] feat(mcp): add persistent client channels --- apps/cloud/src/env-augment.d.ts | 3 + apps/cloud/src/mcp/agent-handler.ts | 38 +- apps/cloud/src/mcp/auth-provider.test.ts | 61 +++ apps/cloud/src/mcp/auth-provider.ts | 20 + .../src/mcp/client-channel-durable-object.ts | 258 ++++++++++++ apps/cloud/src/mcp/client-channel-route.ts | 84 ++++ apps/cloud/src/mcp/client-channel.test.ts | 209 ++++++++++ apps/cloud/src/mcp/client-channel.ts | 220 ++++++++++ apps/cloud/src/mcp/index.ts | 1 + apps/cloud/src/mcp/session-durable-object.ts | 33 ++ apps/cloud/src/mcp/session-meta.node.test.ts | 19 + apps/cloud/src/mcp/session-meta.ts | 4 + apps/cloud/src/server.ts | 6 + apps/cloud/wrangler.jsonc | 8 + apps/docs/docs.json | 2 +- apps/docs/mcp-client-channels.mdx | 142 +++++++ apps/host-selfhost/src/app.ts | 10 + apps/host-selfhost/src/mcp/auth.ts | 287 +++++++------ .../src/mcp/client-binding.test.ts | 87 ++++ .../src/mcp/client-channel-manifest.test.ts | 52 +++ .../src/mcp/client-channel-origin.test.ts | 45 +++ apps/host-selfhost/src/mcp/client-channel.ts | 176 ++++++++ apps/host-selfhost/src/mcp/index.ts | 9 +- apps/host-selfhost/src/mcp/mcp-oauth.test.ts | 10 + apps/host-selfhost/src/mcp/session-store.ts | 3 + e2e/cloud/mcp-client-channel.test.ts | 326 +++++++++++++++ packages/core/api/src/server/mcp-build.ts | 6 + .../core/api/src/server/scoped-executor.ts | 4 + .../mcp/agent-session-durable-object.test.ts | 55 ++- .../src/mcp/agent-session-durable-object.ts | 20 +- .../src/mcp/execution-owner-directory.ts | 8 + .../hosts/cloudflare/src/mcp/session-stub.ts | 3 +- packages/hosts/mcp/package.json | 4 + packages/hosts/mcp/src/envelope.test.ts | 41 ++ packages/hosts/mcp/src/envelope.ts | 6 +- .../mcp/src/in-memory-client-bridge.test.ts | 207 ++++++++++ .../hosts/mcp/src/in-memory-client-bridge.ts | 377 ++++++++++++++++++ .../mcp/src/in-memory-session-store.test.ts | 244 ++++++++++++ .../hosts/mcp/src/in-memory-session-store.ts | 101 ++++- packages/hosts/mcp/src/seams.ts | 18 +- packages/hosts/mcp/src/tool-server.test.ts | 246 +++++++++++- packages/hosts/mcp/src/tool-server.ts | 192 +++++++++ 42 files changed, 3491 insertions(+), 154 deletions(-) create mode 100644 apps/cloud/src/mcp/client-channel-durable-object.ts create mode 100644 apps/cloud/src/mcp/client-channel-route.ts create mode 100644 apps/cloud/src/mcp/client-channel.test.ts create mode 100644 apps/cloud/src/mcp/client-channel.ts create mode 100644 apps/docs/mcp-client-channels.mdx create mode 100644 apps/host-selfhost/src/mcp/client-binding.test.ts create mode 100644 apps/host-selfhost/src/mcp/client-channel-manifest.test.ts create mode 100644 apps/host-selfhost/src/mcp/client-channel-origin.test.ts create mode 100644 apps/host-selfhost/src/mcp/client-channel.ts create mode 100644 e2e/cloud/mcp-client-channel.test.ts create mode 100644 packages/hosts/mcp/src/in-memory-client-bridge.test.ts create mode 100644 packages/hosts/mcp/src/in-memory-client-bridge.ts diff --git a/apps/cloud/src/env-augment.d.ts b/apps/cloud/src/env-augment.d.ts index 3e1fb0d8c1..660986f03d 100644 --- a/apps/cloud/src/env-augment.d.ts +++ b/apps/cloud/src/env-augment.d.ts @@ -5,6 +5,9 @@ declare global { namespace Cloudflare { interface Env { + MCP_CLIENT_CHANNEL: DurableObjectNamespace< + import("./mcp/client-channel-durable-object").McpClientChannelDO + >; /** TOTP enrollment and verification attempts; absence refuses verification. */ ADMIN_MFA_RATE_LIMITER?: { readonly limit: (options: { diff --git a/apps/cloud/src/mcp/agent-handler.ts b/apps/cloud/src/mcp/agent-handler.ts index d10233953a..5cb784e360 100644 --- a/apps/cloud/src/mcp/agent-handler.ts +++ b/apps/cloud/src/mcp/agent-handler.ts @@ -30,7 +30,7 @@ import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { wrapMcpSseResponse } from "../observability/memory-metrics"; import { WorkerTelemetryLive } from "../observability/telemetry"; -import { cloudMcpAuth } from "./auth-provider"; +import { authenticateClientChannelToken, cloudMcpAuth } from "./auth-provider"; import { isMcpSessionMetaUnavailable } from "./session-meta"; import { McpSessionDOSqlite } from "./session-durable-object"; import { parseTraceparent } from "./traceparent"; @@ -44,7 +44,7 @@ const corsPreflightResponse = (): Response => "access-control-allow-origin": "*", "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", "access-control-allow-headers": - "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", + "content-type, authorization, mcp-session-id, accept, mcp-protocol-version, x-executor-client-token", "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", }, }); @@ -171,6 +171,8 @@ const resourceFromPath = (request: Request): McpResource => { const propsForPrincipal = ( request: Request, principal: Extract["principal"], + clientIdentity: string | undefined, + clientManifestGeneration: number | undefined, resource: McpResource, ): Effect.Effect => Effect.gen(function* () { @@ -187,6 +189,8 @@ const propsForPrincipal = ( ...(principal.organizationSlug ? { organizationSlug: principal.organizationSlug } : {}), ...sessionOrgRoleMetadata(principal), userId: principal.accountId, + ...(clientIdentity === undefined ? {} : { clientIdentity }), + ...(clientManifestGeneration === undefined ? {} : { clientManifestGeneration }), elicitationMode: readElicitationMode(request), artifactsEnabled: readArtifactsEnabled(request), searchToolsEnabled: readSearchToolsEnabled(request), @@ -246,6 +250,23 @@ export const makeCloudMcpAgentHandler = () => { return renderAuthError(auth, request, outcome); } + // A channel token is only a session binding layered on top of the normal + // cloud bearer auth above. Its opaque value selects a DO; identity is read + // from durable server-side state and must match the bearer principal. + const channelToken = request.headers.get("x-executor-client-token"); + let clientIdentity = outcome.clientIdentity; + let clientManifestGeneration = outcome.clientManifestGeneration; + if (channelToken !== null) { + const channelIdentity = await authenticateClientChannelToken( + env.MCP_CLIENT_CHANNEL, + channelToken, + outcome.principal, + ); + if (!channelIdentity) return jsonRpcResponse(401, -32001, "Invalid client channel token"); + clientIdentity = channelIdentity.binding; + clientManifestGeneration = channelIdentity.generation; + } + if (!sessionId && request.method === "DELETE") { // Matches the old envelope's contract (@modelcontextprotocol/sdk's // `WebStandardStreamableHTTPServerTransport.handleDeleteRequest`): 200, @@ -266,6 +287,8 @@ export const makeCloudMcpAgentHandler = () => { { accountId: outcome.principal.accountId, organizationId: outcome.principal.organizationId, + ...(clientIdentity === undefined ? {} : { clientIdentity }), + ...(clientManifestGeneration === undefined ? {} : { clientManifestGeneration }), }, resource, ); @@ -297,7 +320,16 @@ export const makeCloudMcpAgentHandler = () => { } } - const props = await runTraced(request, propsForPrincipal(request, outcome.principal, resource)); + const props = await runTraced( + request, + propsForPrincipal( + request, + outcome.principal, + clientIdentity, + clientManifestGeneration, + resource, + ), + ); (ctx as ExecutionContext & { props?: McpSessionProps }).props = props; const forwarded = withOrgWriteAccess( withVerifiedIdentityHeaders( diff --git a/apps/cloud/src/mcp/auth-provider.test.ts b/apps/cloud/src/mcp/auth-provider.test.ts index 1a88793d69..50500c5b39 100644 --- a/apps/cloud/src/mcp/auth-provider.test.ts +++ b/apps/cloud/src/mcp/auth-provider.test.ts @@ -26,6 +26,7 @@ import { McpAuthProvider, orgWriteAccessForPrincipal } from "@executor-js/host-m import { WorkOSError } from "../auth/errors"; import { cloudMcpAuthProviderLayer } from "./auth-provider"; +import { authenticateClientChannelToken } from "./auth-provider"; import { MCP_ORGANIZATION_HEADER, McpAuth, @@ -33,6 +34,7 @@ import { mcpAuthorized, mcpUnauthorized, } from "./auth"; +import { authenticateClientToken, createClientChannelState } from "./client-channel"; const ACCOUNT_ID = "user_test"; const ORG_ID = "org_test"; @@ -96,6 +98,65 @@ const authenticateWith = ( ); describe("cloud MCP org-authorization classification", () => { + it("resolves channel credentials to durable identity and generation, rejecting invalid tokens", () => { + const state = createClientChannelState(); + const binding = state.register( + { accountId: ACCOUNT_ID, organizationId: ORG_ID }, + { + tools: [{ name: "alpha" }], + resources: [], + prompts: [], + }, + ); + expect(typeof binding).toBe("string"); + if (typeof binding !== "string") return; + expect(authenticateClientToken(state, binding)).toEqual({ + principal: { accountId: ACCOUNT_ID, organizationId: ORG_ID }, + generation: 0, + binding, + }); + expect(authenticateClientToken(state, "forged-token")).toBeNull(); + state.disconnect(binding); + expect(authenticateClientToken(state, binding)).toBeNull(); + }); + + it("authenticates supplied channel tokens against verified tenant identity", async () => { + const state = createClientChannelState(); + const binding = state.register( + { accountId: ACCOUNT_ID, organizationId: ORG_ID }, + { + tools: [{ name: "alpha" }], + resources: [], + prompts: [], + }, + ); + expect(typeof binding).toBe("string"); + if (typeof binding !== "string") return; + const namespace = { + idFromName: (name: string) => name, + get: () => ({ + authenticate: (token: string) => Promise.resolve(authenticateClientToken(state, token)), + }), + }; + await expect( + authenticateClientChannelToken(namespace, binding, { + accountId: ACCOUNT_ID, + organizationId: ORG_ID, + }), + ).resolves.toMatchObject({ binding, generation: 0 }); + await expect( + authenticateClientChannelToken(namespace, "invalid-token", { + accountId: ACCOUNT_ID, + organizationId: ORG_ID, + }), + ).resolves.toBeNull(); + await expect( + authenticateClientChannelToken(namespace, binding, { + accountId: "other", + organizationId: ORG_ID, + }), + ).resolves.toBeNull(); + }); it.effect("WorkOS 5xx -> Unavailable (retryable 503, session preserved)", () => Effect.gen(function* () { const outcome = yield* authenticateWith(stubOrgAuthFailing(new WorkOSError({ status: 503 }))); diff --git a/apps/cloud/src/mcp/auth-provider.ts b/apps/cloud/src/mcp/auth-provider.ts index b05685af58..e79d3e6d6c 100644 --- a/apps/cloud/src/mcp/auth-provider.ts +++ b/apps/cloud/src/mcp/auth-provider.ts @@ -67,6 +67,26 @@ import { authorizationServerMetadataResponse, protectedResourceMetadataResponse, } from "./oauth-metadata"; +import type { McpClientChannelDO } from "./client-channel-durable-object"; +import { clientChannelBelongsToPrincipal } from "./client-channel"; +import { mcpClientChannelIdFromName } from "@executor-js/cloudflare/mcp/execution-owner-directory"; + +export interface McpClientChannelNamespace { + readonly idFromName: (name: string) => Id; + readonly get: (id: Id) => Pick; +} + +export const authenticateClientChannelToken = async ( + namespace: McpClientChannelNamespace, + token: string, + principal: Pick, +) => { + if (!/^[0-9a-f-]{36}$/i.test(token)) return null; + const channel = namespace.get(mcpClientChannelIdFromName(namespace, token)); + const identity = await channel.authenticate(token); + if (!clientChannelBelongsToPrincipal(identity, principal)) return null; + return identity; +}; const AUTHORIZATION_SERVER_METADATA_PATH = "/.well-known/oauth-authorization-server"; const TOOLKIT_PROTECTED_RESOURCE_METADATA_PATH = `${PROTECTED_RESOURCE_METADATA_PATH}/toolkits/:toolkitSlug`; diff --git a/apps/cloud/src/mcp/client-channel-durable-object.ts b/apps/cloud/src/mcp/client-channel-durable-object.ts new file mode 100644 index 0000000000..cf0e7b6e44 --- /dev/null +++ b/apps/cloud/src/mcp/client-channel-durable-object.ts @@ -0,0 +1,258 @@ +import { DurableObject } from "cloudflare:workers"; +import { mcpClientChannelIdFromName } from "@executor-js/cloudflare/mcp/execution-owner-directory"; +import { Data, Effect, Exit, Option, Schema } from "effect"; + +import { + createClientChannelState, + filterClientManifestResponse, + routeClientRequest, + type ClientChannelSnapshot, + type ClientManifest, + type ClientPrincipal, +} from "./client-channel"; + +export type PendingClientInvocation = { + readonly resume: (effect: Effect.Effect) => void; + readonly timeout?: ReturnType; +}; + +class McpClientChannelError extends Data.TaggedError("McpClientChannelError")<{ + readonly reason: string; +}> { + override get message(): string { + return this.reason; + } +} + +const decodeClientFrame = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); + +export const dispatchClientInvocation = ( + pending: Map, + id: string, + socket: Pick, + method: string, + params: unknown, +): Promise => + Effect.runPromise( + Effect.callback((resume) => { + const timeout = setTimeout(() => { + if (pending.delete(id)) + resume(Effect.fail(new McpClientChannelError({ reason: "Client invocation timed out" }))); + }, 30_000); + pending.set(id, { resume, timeout }); + const sent = Effect.runSyncExit( + Effect.try({ + try: () => socket.send(JSON.stringify({ id, method, params })), + catch: () => new McpClientChannelError({ reason: "Client invocation delivery failed" }), + }), + ); + if (Exit.isFailure(sent)) { + pending.delete(id); + clearTimeout(timeout); + resume( + Effect.fail(new McpClientChannelError({ reason: "Client invocation delivery failed" })), + ); + } + return Effect.sync(() => { + pending.delete(id); + clearTimeout(timeout); + }); + }), + ); + +export const resolveClientInvocationResponse = ( + pending: Map, + binding: string, + frame: unknown, +): boolean => { + if ( + typeof frame !== "object" || + frame === null || + !("id" in frame) || + typeof frame.id !== "string" || + !frame.id.startsWith(`${binding}:`) + ) + return false; + const request = pending.get(frame.id); + if (!request) return false; + pending.delete(frame.id); + if (request.timeout !== undefined) clearTimeout(request.timeout); + if ("error" in frame) + request.resume(Effect.fail(new McpClientChannelError({ reason: "Client invocation failed" }))); + else if ("result" in frame) request.resume(Effect.succeed(frame.result)); + else + request.resume( + Effect.fail(new McpClientChannelError({ reason: "Invalid client invocation response" })), + ); + return true; +}; + +/** Per-client durable router. All identity inputs must originate from the + * Worker after cloudMcpAuth succeeds; client messages never supply identity. */ +export class McpClientChannelDO extends DurableObject { + private statePromise: Promise>; + private readonly pending = new Map(); + private sequence = 0; + + constructor(ctx: DurableObjectState, env: Env) { + super(ctx, env); + this.statePromise = ctx.blockConcurrencyWhile(async () => { + const snapshot = await ctx.storage.get("clients"); + return createClientChannelState(snapshot); + }); + } + + async register(principal: ClientPrincipal, manifest: ClientManifest, clientId: string) { + const state = await this.statePromise; + const binding = state.register(principal, manifest, clientId); + if (typeof binding === "string") await this.ctx.storage.put("clients", state.snapshot()); + return binding; + } + + async route(binding: string, generation: number, request: unknown) { + const state = await this.statePromise; + return routeClientRequest(state, binding, generation, request); + } + + async filterResponse(binding: string, method: string, response: unknown) { + const state = await this.statePromise; + return filterClientManifestResponse(state, binding, method, response); + } + + async getManifest(binding: string) { + const state = await this.statePromise; + const entry = state.get(binding); + return entry?.connected ? entry.manifest : null; + } + + async authenticate(token: string | null) { + const state = await this.statePromise; + if (!token) return null; + const entry = state.get(token); + return entry?.connected + ? { binding: token, generation: entry.generation, principal: entry.principal } + : null; + } + + async revoke(binding: string) { + const state = await this.statePromise; + state.disconnect(binding); + await this.ctx.storage.put("clients", state.snapshot()); + } + + async fetch(request: Request): Promise { + if (request.headers.get("upgrade") !== "websocket") + return new Response("Upgrade required", { status: 426 }); + const token = request.headers.get("x-executor-client-token"); + const identity = await this.authenticate(token); + if (!identity) return new Response("Unauthorized", { status: 401 }); + const pair = new WebSocketPair(); + const [client, server] = Object.values(pair) as [WebSocket, WebSocket]; + this.ctx.acceptWebSocket(server); + server.serializeAttachment({ + binding: identity.binding, + generation: identity.generation, + }); + return new Response(null, { status: 101, webSocket: client }); + } + + async webSocketMessage(socket: WebSocket, message: string | ArrayBuffer) { + const body = typeof message === "string" ? message : new TextDecoder().decode(message); + if (body.length > 65_536) { + socket.close(1009, "Message too large"); + return; + } + const attachmentExit = Effect.runSyncExit( + Effect.try({ + try: () => + socket.deserializeAttachment() as { + binding?: unknown; + generation?: unknown; + } | null, + catch: () => new McpClientChannelError({ reason: "Invalid client attachment" }), + }), + ); + if (Exit.isFailure(attachmentExit)) { + socket.close(1011, "Routing failed"); + return; + } + const attachment = attachmentExit.value; + if ( + !attachment || + typeof attachment.binding !== "string" || + typeof attachment.generation !== "number" + ) { + socket.close(1008, "Invalid client binding"); + return; + } + const parsed = decodeClientFrame(body); + if (Option.isNone(parsed)) { + socket.close(1008, "Invalid client response"); + return; + } + if (resolveClientInvocationResponse(this.pending, attachment.binding, parsed.value)) return; + socket.close(1008, "Uncorrelated client message"); + } + async webSocketClose(socket: WebSocket) { + const attachment = socket.deserializeAttachment() as { binding?: unknown } | null; + if (typeof attachment?.binding === "string") { + for (const [id, pending] of this.pending) { + if (id.startsWith(`${attachment.binding}:`)) { + this.pending.delete(id); + if (pending.timeout !== undefined) clearTimeout(pending.timeout); + pending.resume(Effect.fail(new McpClientChannelError({ reason: "Client disconnected" }))); + } + } + await this.revoke(attachment.binding); + } + } + + async invoke( + binding: string, + generation: number, + method: string, + params: unknown, + ): Promise { + const state = await this.statePromise; + const validation = routeClientRequest(state, binding, generation, { method, params }); + if (!validation.ok) + return Effect.runPromise( + Effect.fail( + new McpClientChannelError({ reason: validation.error ?? "Client request denied" }), + ), + ); + const socket = this.ctx.getWebSockets().find((candidate) => { + const attachment = candidate.deserializeAttachment() as { + binding?: unknown; + generation?: unknown; + } | null; + return attachment?.binding === binding && attachment.generation === generation; + }); + if (!socket) + return Effect.runPromise( + Effect.fail(new McpClientChannelError({ reason: "Client is disconnected" })), + ); + const id = `${binding}:${++this.sequence}`; + return dispatchClientInvocation(this.pending, id, socket, method, params); + } +} + +/** Stable adapter for MCP session builders; caller supplies identity obtained + * only from the verified cloud auth outcome. */ +export const authenticatedClientChannel = ( + namespace: DurableObjectNamespace, + principal: ClientPrincipal, + manifest: ClientManifest, +) => { + const clientId = crypto.randomUUID(); + const stub = namespace.get(mcpClientChannelIdFromName(namespace, clientId)); + return { + clientId, + register: () => stub.register(principal, manifest, clientId), + route: (generation: number, request: unknown) => stub.route(clientId, generation, request), + filterResponse: (method: string, response: unknown) => + stub.filterResponse(clientId, method, response), + revoke: () => stub.revoke(clientId), + authenticate: (token: string) => stub.authenticate(token), + }; +}; diff --git a/apps/cloud/src/mcp/client-channel-route.ts b/apps/cloud/src/mcp/client-channel-route.ts new file mode 100644 index 0000000000..737c31ecaf --- /dev/null +++ b/apps/cloud/src/mcp/client-channel-route.ts @@ -0,0 +1,84 @@ +import { Effect, Option, Predicate, Schema } from "effect"; + +import { mcpClientChannelIdFromName } from "@executor-js/cloudflare/mcp/execution-owner-directory"; +import { McpAuthProvider } from "@executor-js/host-mcp"; + +import { cloudMcpAuth } from "./auth-provider"; +import { isClientManifest } from "./client-channel"; + +const json = (value: unknown, status = 200) => + Response.json(value, { status, headers: { "cache-control": "no-store" } }); +const ClientManifestEnvelopeSchema = Schema.Struct({ manifest: Schema.Unknown }); +const decodeClientManifestEnvelope = Schema.decodeUnknownOption( + Schema.fromJsonString(ClientManifestEnvelopeSchema), +); + +/** Worker-owned entry point. Registration always authenticates the ordinary + * cloud bearer; client-supplied identity fields are never consumed. */ +export const handleMcpClientChannel = async (request: Request, env: Env): Promise => { + const outcome = await Effect.runPromise( + Effect.gen(function* () { + const provider = yield* McpAuthProvider; + return yield* provider.authenticate(request); + }).pipe(Effect.provide(cloudMcpAuth)), + ); + if (!Predicate.isTagged(outcome, "Authenticated")) { + const status = Predicate.isTagged(outcome, "Unauthorized") + ? 401 + : Predicate.isTagged(outcome, "Forbidden") + ? 403 + : 503; + return json( + { error: status === 401 ? "Unauthorized" : status === 403 ? "Forbidden" : "Unavailable" }, + status, + ); + } + const token = request.headers.get("x-executor-client-token"); + if (request.headers.get("upgrade")?.toLowerCase() === "websocket") { + if (!token) return json({ error: "Client channel token required" }, 401); + const stub = env.MCP_CLIENT_CHANNEL.get( + mcpClientChannelIdFromName(env.MCP_CLIENT_CHANNEL, token), + ); + const identity = await stub.authenticate(token); + if ( + !identity || + identity.principal.accountId !== outcome.principal.accountId || + identity.principal.organizationId !== outcome.principal.organizationId + ) + return json({ error: "Invalid client channel token" }, 401); + return stub.fetch(request); + } + if (request.method !== "POST") return json({ error: "Method not allowed" }, 405); + const binding = crypto.randomUUID(); + const stub = env.MCP_CLIENT_CHANNEL.get( + mcpClientChannelIdFromName(env.MCP_CLIENT_CHANNEL, binding), + ); + if (Number(request.headers.get("content-length")) > 262_144) { + return json({ error: "Client manifest too large" }, 413); + } + const rawBody = await request.text(); + if (new TextEncoder().encode(rawBody).byteLength > 262_144) { + return json({ error: "Client manifest too large" }, 413); + } + const decoded = decodeClientManifestEnvelope(rawBody); + if (Option.isNone(decoded)) return json({ error: "Invalid client manifest" }, 400); + const manifest = decoded.value.manifest; + if (!isClientManifest(manifest)) return json({ error: "Invalid client manifest" }, 400); + const registered = await stub.register( + { + accountId: outcome.principal.accountId, + organizationId: outcome.principal.organizationId, + }, + manifest, + binding, + ); + if (typeof registered !== "string") return json({ error: "Invalid client manifest" }, 400); + return json( + { + token: registered, + websocket: `${new URL(request.url).origin}/mcp/client-channel`, + generation: 0, + }, + 201, + ); +}; diff --git a/apps/cloud/src/mcp/client-channel.test.ts b/apps/cloud/src/mcp/client-channel.test.ts new file mode 100644 index 0000000000..97da518220 --- /dev/null +++ b/apps/cloud/src/mcp/client-channel.test.ts @@ -0,0 +1,209 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { + createClientChannelState, + clientChannelBelongsToPrincipal, + filterClientManifestResponse, + routeClientRequest, +} from "./client-channel"; +import { + dispatchClientInvocation, + resolveClientInvocationResponse, + type PendingClientInvocation, +} from "./client-channel-durable-object"; + +describe("durable MCP client routing", () => { + it("clears pending invocation state when WebSocket send throws", async () => { + const pending = new Map(); + await expect( + dispatchClientInvocation( + pending, + "binding-a:1", + { + send: () => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- test boundary: the WebSocket adapter must exercise its synchronous send-failure path + throw new Error("socket is closed"); + }, + }, + "tools/call", + { name: "tool" }, + ), + ).rejects.toThrow("Client invocation delivery failed"); + expect(pending.size).toBe(0); + }); + + it("correlates client responses only to their server-issued binding and request", async () => { + let resolveResult: (value: unknown) => void = () => {}; + let rejectResult: (reason: Error) => void = () => {}; + const pending = new Map([ + [ + "binding-a:1", + { + resume: (effect: Effect.Effect) => { + void Effect.runPromise(effect).then(resolveResult, rejectResult); + }, + }, + ], + ]); + const result = new Promise((resolve, reject) => { + resolveResult = resolve; + rejectResult = reject; + }); + expect( + resolveClientInvocationResponse(pending, "binding-b", { + id: "binding-a:1", + result: { content: [] }, + }), + ).toBe(false); + expect( + resolveClientInvocationResponse(pending, "binding-a", { + id: "binding-a:1", + result: { content: [{ type: "text", text: "response" }] }, + }), + ).toBe(true); + await expect(result).resolves.toEqual({ + content: [{ type: "text", text: "response" }], + }); + expect(pending.size).toBe(0); + }); + + it("isolates A/B same-name capabilities and standard list keys", () => { + const state = createClientChannelState(); + const clientA = state.register( + { accountId: "u", organizationId: "org" }, + { + tools: [{ name: "shared", description: "A" }], + resources: [{ uri: "client://a" }], + prompts: [{ name: "prompt-a" }], + }, + ); + const clientB = state.register( + { accountId: "u", organizationId: "org" }, + { + tools: [{ name: "shared", description: "B" }], + resources: [{ uri: "client://b" }], + prompts: [{ name: "prompt-b" }], + }, + ); + expect(typeof clientA).toBe("string"); + expect(typeof clientB).toBe("string"); + if (typeof clientA !== "string" || typeof clientB !== "string") return; + expect(clientA).not.toBe(clientB); + expect( + routeClientRequest(state, clientA, 0, { + method: "tools/call", + params: { name: "shared" }, + }), + ).toMatchObject({ ok: true }); + expect( + routeClientRequest(state, clientA, 0, { + method: "resources/read", + params: { uri: "client://b" }, + }), + ).toMatchObject({ ok: false, error: "manifest_denied" }); + expect( + routeClientRequest(state, clientA, 0, { + method: "prompts/get", + params: { name: "prompt-b" }, + }), + ).toMatchObject({ ok: false, error: "manifest_denied" }); + expect( + filterClientManifestResponse(state, clientA, "tools/list", { + result: { tools: [{ name: "shared", description: "A" }, { name: "other" }] }, + }), + ).toEqual({ result: { tools: [{ name: "shared", description: "A" }] } }); + }); + + it("denies cross-tenant routing and stale generations after restore", () => { + const state = createClientChannelState(); + const binding = state.register( + { accountId: "u", organizationId: "org" }, + { + tools: [], + resources: [{ uri: "file://secret" }], + prompts: [{ name: "alpha" }], + }, + ); + expect(typeof binding).toBe("string"); + if (typeof binding !== "string") return; + const restored = createClientChannelState(state.snapshot()); + expect( + routeClientRequest(restored, binding, 0, { + method: "resources/read", + params: { uri: "file://secret" }, + principal: { accountId: "forged", organizationId: "org" }, + }), + ).toMatchObject({ ok: false, error: "client_identity_forbidden" }); + expect( + routeClientRequest(restored, binding, 99, { + method: "tools/list", + }), + ).toMatchObject({ ok: false }); + expect( + routeClientRequest(restored, binding, 0, { + method: "prompts/get", + params: { name: "alpha" }, + }), + ).toMatchObject({ ok: true }); + }); + + it("filters standard resource and prompt list results", () => { + const state = createClientChannelState(); + const binding = state.register( + { accountId: "u", organizationId: "org" }, + { + tools: [], + resources: [{ uri: "client://one" }], + prompts: [{ name: "client_prompt" }], + }, + ); + expect(typeof binding).toBe("string"); + if (typeof binding !== "string") return; + expect( + filterClientManifestResponse(state, binding, "resources/list", { + result: { resources: [{ uri: "client://one" }, { uri: "private://two" }] }, + }), + ).toEqual({ result: { resources: [{ uri: "client://one" }] } }); + expect( + filterClientManifestResponse(state, binding, "prompts/list", { + result: { prompts: [{ name: "client_prompt" }, { name: "private_prompt" }] }, + }), + ).toEqual({ result: { prompts: [{ name: "client_prompt" }] } }); + }); + + it("requires a newly issued binding after reconnect", () => { + const state = createClientChannelState(); + const manifest = { tools: [{ name: "alpha" }], resources: [], prompts: [] }; + const oldBinding = state.register({ accountId: "u", organizationId: "org" }, manifest); + expect(typeof oldBinding).toBe("string"); + if (typeof oldBinding !== "string") return; + state.disconnect(oldBinding); + const newBinding = state.register({ accountId: "u", organizationId: "org" }, manifest); + expect(newBinding).not.toBe(oldBinding); + expect(routeClientRequest(state, oldBinding, 0, { method: "tools/list" })).toMatchObject({ + ok: false, + }); + }); + + it("rejects a binding authenticated for another tenant", () => { + const state = createClientChannelState(); + const binding = state.register( + { accountId: "user-a", organizationId: "org-a" }, + { + tools: [], + resources: [], + prompts: [], + }, + ); + expect(typeof binding).toBe("string"); + if (typeof binding !== "string") return; + const identity = state.get(binding) ?? null; + expect( + clientChannelBelongsToPrincipal(identity, { + accountId: "user-b", + organizationId: "org-a", + }), + ).toBe(false); + }); +}); diff --git a/apps/cloud/src/mcp/client-channel.ts b/apps/cloud/src/mcp/client-channel.ts new file mode 100644 index 0000000000..a9c8f4120e --- /dev/null +++ b/apps/cloud/src/mcp/client-channel.ts @@ -0,0 +1,220 @@ +import { Option, Schema } from "effect"; + +export type ClientPrincipal = { + readonly accountId: string; + readonly organizationId: string; +}; +export type ClientManifest = { + readonly tools: readonly { readonly name: string; readonly [key: string]: unknown }[]; + readonly resources: readonly { readonly uri: string; readonly [key: string]: unknown }[]; + readonly prompts: readonly { readonly name: string; readonly [key: string]: unknown }[]; +}; +export const isClientManifest = (value: unknown): value is ClientManifest => { + if (typeof value !== "object" || value === null) return false; + if (!("tools" in value) || !Array.isArray(value.tools)) return false; + if (!("resources" in value) || !Array.isArray(value.resources)) return false; + if (!("prompts" in value) || !Array.isArray(value.prompts)) return false; + return ( + value.tools.every((item) => typeof item === "object" && item !== null && "name" in item) && + value.resources.every((item) => typeof item === "object" && item !== null && "uri" in item) && + value.prompts.every((item) => typeof item === "object" && item !== null && "name" in item) + ); +}; +export type ClientBinding = string; +export type ClientChannelEntry = { + readonly principal: ClientPrincipal; + readonly manifest: ClientManifest; + readonly generation: number; + readonly connected: boolean; +}; +export type ClientChannelSnapshot = Readonly>; + +const RequestSchema = Schema.Struct({ + method: Schema.String, + params: Schema.optional(Schema.Unknown), +}); +const decodeRequest = Schema.decodeUnknownOption(RequestSchema); + +export const createClientChannelState = (initial: ClientChannelSnapshot = {}) => { + const entries = new Map(Object.entries(initial)); + return { + register( + principal: ClientPrincipal, + manifest: ClientManifest, + clientId: string = crypto.randomUUID(), + ): ClientBinding | { ok: false } { + if ( + !principal?.accountId || + !principal?.organizationId || + !manifest || + !Array.isArray(manifest.tools) || + !Array.isArray(manifest.resources) || + !Array.isArray(manifest.prompts) || + manifest.tools.length + manifest.resources.length + manifest.prompts.length > 100 + ) + return { ok: false }; + const allItems = [...manifest.tools, ...manifest.resources, ...manifest.prompts]; + if (allItems.some((item) => typeof item !== "object" || item === null)) return { ok: false }; + const uniqueNames = [ + manifest.tools.map((item) => item.name), + manifest.resources.map((item) => item.uri), + manifest.prompts.map((item) => item.name), + ]; + if ( + uniqueNames.some( + (names) => + names.some((name) => typeof name !== "string" || !name || name.length > 200) || + new Set(names).size !== names.length, + ) + ) + return { ok: false }; + if (entries.has(clientId)) return { ok: false }; + const binding = clientId; + entries.set(binding, { + principal, + manifest: { + tools: [...manifest.tools], + resources: [...manifest.resources], + prompts: [...manifest.prompts], + }, + generation: 0, + connected: true, + }); + return binding; + }, + snapshot(): ClientChannelSnapshot { + return Object.fromEntries(entries); + }, + get(binding: string) { + return entries.get(binding); + }, + disconnect(binding: string) { + const entry = entries.get(binding); + if (entry) + entries.set(binding, { ...entry, connected: false, generation: entry.generation + 1 }); + }, + }; +}; + +export const routeClientRequest = ( + state: ReturnType, + binding: string, + generation: number, + raw: unknown, +): { readonly ok: boolean; readonly error?: string } => { + if (JSON.stringify(raw ?? null).length > 65_536) return { ok: false, error: "request_too_large" }; + if (typeof raw === "object" && raw !== null && "principal" in raw) { + return { ok: false, error: "client_identity_forbidden" }; + } + const parsed = decodeRequest(raw); + if (Option.isNone(parsed)) return { ok: false, error: "invalid_request" }; + const request = parsed.value; + const entry = state.get(binding); + if (!entry || !entry.connected || entry.generation !== generation) + return { ok: false, error: "stale_client" }; + const params = request.params; + const selector = + typeof params === "object" && params !== null + ? "name" in params && typeof params.name === "string" + ? params.name + : "uri" in params && typeof params.uri === "string" + ? params.uri + : undefined + : undefined; + const manifestMethods: Readonly> = { + "tools/call": "name", + "resources/read": "uri", + "prompts/get": "name", + }; + if ( + !(request.method in manifestMethods) && + !["tools/list", "resources/list", "prompts/list"].includes(request.method) + ) { + return { ok: false, error: "method_denied" }; + } + const allowedSelectors = + request.method === "tools/call" + ? entry.manifest.tools.map((x) => x.name) + : request.method === "resources/read" + ? entry.manifest.resources.map((x) => x.uri) + : request.method === "prompts/get" + ? entry.manifest.prompts.map((x) => x.name) + : []; + if ( + request.method in manifestMethods && + (selector === undefined || !allowedSelectors.includes(selector)) + ) { + return { ok: false, error: "manifest_denied" }; + } + return { ok: true }; +}; + +/** Resolve an opaque, server-issued channel credential. Never decode identity + * from the credential: the durable record is the source of truth. */ +export const authenticateClientToken = ( + state: ReturnType, + token: string | null, +): { + readonly principal: ClientPrincipal; + readonly generation: number; + readonly binding: string; +} | null => { + if (!token || token.length > 128) return null; + const entry = state.get(token); + return entry?.connected + ? { principal: entry.principal, generation: entry.generation, binding: token } + : null; +}; + +export const clientChannelBelongsToPrincipal = ( + identity: { readonly principal: ClientPrincipal } | null, + principal: ClientPrincipal, +): boolean => + identity !== null && + identity.principal.accountId === principal.accountId && + identity.principal.organizationId === principal.organizationId; + +/** Filter standard MCP list-result arrays to the persisted manifest before + * returning them through an overlay/session bridge. Unknown response shapes + * fail closed rather than leaking an unfiltered catalog. */ +export const filterClientManifestResponse = ( + state: ReturnType, + binding: string, + method: string, + raw: unknown, +): unknown | null => { + const entry = state.get(binding); + if (!entry?.connected) return null; + if (!["tools/list", "resources/list", "prompts/list"].includes(method)) return null; + if ( + typeof raw !== "object" || + raw === null || + !("result" in raw) || + typeof raw.result !== "object" || + raw.result === null + ) + return null; + const result = raw.result as Record; + const key = + method === "tools/list" ? "tools" : method === "resources/list" ? "resources" : "prompts"; + if (!(key in result) || !Array.isArray(result[key])) return null; + const selector = key === "resources" ? "uri" : "name"; + const allowed = new Set( + key === "tools" + ? entry.manifest.tools.map((x) => x.name) + : key === "resources" + ? entry.manifest.resources.map((x) => x.uri) + : entry.manifest.prompts.map((x) => x.name), + ); + return { + ...raw, + result: { + ...result, + [key]: result[key].filter((item: unknown) => { + if (typeof item !== "object" || item === null) return false; + const itemSelector = (item as Record)[selector]; + return typeof itemSelector === "string" && allowed.has(itemSelector); + }), + }, + }; +}; diff --git a/apps/cloud/src/mcp/index.ts b/apps/cloud/src/mcp/index.ts index a7ec769364..3939d0a23f 100644 --- a/apps/cloud/src/mcp/index.ts +++ b/apps/cloud/src/mcp/index.ts @@ -17,3 +17,4 @@ // its `McpAuth`/`McpOrganizationAuth` seams provided internally), shaped as the // `Layer` `ExecutorApp.make` expects. export { cloudMcpAuth } from "./auth-provider"; +export { authenticatedClientChannel } from "./client-channel-durable-object"; diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index 81d9b50d4e..bd156fe6c3 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -44,6 +44,8 @@ import { type McpExecutionOwnerRoute, } from "@executor-js/cloudflare/mcp/execution-owner-directory"; import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; +import { mcpClientChannelIdFromName } from "@executor-js/cloudflare/mcp/execution-owner-directory"; +import type { ClientManifest } from "./client-channel"; import { buildExecuteDescription, parseIntegrationInventory, @@ -350,6 +352,12 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase clientChannel.getManifest(clientBinding), + catch: () => null, + }).pipe(Effect.orElseSucceed(() => null)) + : null; const mcpServer = yield* createExecutorMcpServer({ engine, description, @@ -413,6 +434,18 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase + clientChannel.invoke( + clientBinding, + sessionMeta.clientManifestGeneration ?? 0, + request.method, + request.params, + ), + } + : {}), elicitationMode: sessionElicitationMode === "browser" ? { diff --git a/apps/cloud/src/mcp/session-meta.node.test.ts b/apps/cloud/src/mcp/session-meta.node.test.ts index f745da3ee5..64243e7fbe 100644 --- a/apps/cloud/src/mcp/session-meta.node.test.ts +++ b/apps/cloud/src/mcp/session-meta.node.test.ts @@ -144,6 +144,25 @@ describe("resolveSessionMetaForToken", () => { expect(store.calls()).toBe(0); }); + it("persists the authenticated client and manifest generation in session metadata", async () => { + const store = countingConnectTimeoutStore(); + + const meta = await Effect.runPromise( + resolveSessionMetaForToken( + { + ...TOKEN, + clientIdentity: "linked-client-a", + clientManifestGeneration: 3, + }, + STORED, + ).pipe(Effect.provide(Layer.mergeAll(store.layer, unusedWorkOS))), + ); + + expect(meta.clientIdentity).toBe("linked-client-a"); + expect(meta.clientManifestGeneration).toBe(3); + expect(store.calls()).toBe(0); + }); + it("reads the database only when nothing else names the org", async () => { const store = namingStore(); diff --git a/apps/cloud/src/mcp/session-meta.ts b/apps/cloud/src/mcp/session-meta.ts index 1dd06a8866..a4936b32f5 100644 --- a/apps/cloud/src/mcp/session-meta.ts +++ b/apps/cloud/src/mcp/session-meta.ts @@ -123,6 +123,10 @@ const metaFromIdentity = ( ...(organization.slug === undefined ? {} : { organizationSlug: organization.slug }), ...sessionOrgRoleMetadata(token), userId: token.userId, + ...(token.clientIdentity === undefined ? {} : { clientIdentity: token.clientIdentity }), + ...(token.clientManifestGeneration === undefined + ? {} + : { clientManifestGeneration: token.clientManifestGeneration }), resource: token.resource, elicitationMode: token.elicitationMode, artifactsEnabled: token.artifactsEnabled, diff --git a/apps/cloud/src/server.ts b/apps/cloud/src/server.ts index ca00c5132b..47c5e40b8a 100644 --- a/apps/cloud/src/server.ts +++ b/apps/cloud/src/server.ts @@ -19,6 +19,8 @@ import { makeCloudMcpAgentHandler } from "./mcp/agent-handler"; import { classifyMcpPath, prepareMcpOrgScope } from "./mcp/mount"; import { parseTraceparent } from "./mcp/traceparent"; import { McpSessionDOSqlite as McpSessionDOBase } from "./mcp/session-durable-object"; +import { McpClientChannelDO } from "./mcp/client-channel-durable-object"; +import { handleMcpClientChannel } from "./mcp/client-channel-route"; import { cloudSentryOptions, captureCause, @@ -48,6 +50,7 @@ export const McpSessionDOSqlite = Sentry.instrumentDurableObjectWithSentry( // exported here only to satisfy the migration. It can be removed in a later deploy // (with a `deleted_classes: ["McpSessionDO"]` migration) now that nothing binds it. export class McpSessionDO extends DurableObject {} +export { McpClientChannelDO }; // Per-org execution rate-limit counter DO (abuse backstop; migration v3, // `EXECUTION_RATE_LIMITER` binding). Plain counter, no Sentry wrapper needed: @@ -349,6 +352,9 @@ const cloudflareHandler = { // early return below: the handler invokes it for uncached MCP traffic, and // this entry invokes it for non-MCP paths. const url = new URL(request.url); + if (url.pathname === "/mcp/client-channel") { + return handleMcpClientChannel(request, env); + } const mcpRoute = classifyMcpPath(url.pathname); if (mcpRoute?.kind === "mcp") { // The Cloudflare Agents MCP bridge needs the platform ExecutionContext diff --git a/apps/cloud/wrangler.jsonc b/apps/cloud/wrangler.jsonc index 21cba7aac1..d4a7c391c4 100644 --- a/apps/cloud/wrangler.jsonc +++ b/apps/cloud/wrangler.jsonc @@ -61,6 +61,10 @@ "name": "EXECUTION_RATE_LIMITER", "class_name": "ExecutionRateLimiterDO", }, + { + "name": "MCP_CLIENT_CHANNEL", + "class_name": "McpClientChannelDO", + }, ], }, // The MCP session DO moved to the Cloudflare Agents (`McpAgent`) base, which @@ -89,6 +93,10 @@ "tag": "v4", "new_sqlite_classes": ["McpExecutionOwnerDirectoryDO"], }, + { + "tag": "v5", + "new_sqlite_classes": ["McpClientChannelDO"], + }, ], "services": [ // v1's marketing worker. On executor.sh it now serves only v1's terms of diff --git a/apps/docs/docs.json b/apps/docs/docs.json index f05ef3ec2f..a57d517cba 100644 --- a/apps/docs/docs.json +++ b/apps/docs/docs.json @@ -25,7 +25,7 @@ "groups": [ { "group": "Get started", - "pages": ["index", "mcp-proxy"] + "pages": ["index", "mcp-proxy", "mcp-client-channels"] }, { "group": "Local", diff --git a/apps/docs/mcp-client-channels.mdx b/apps/docs/mcp-client-channels.mdx new file mode 100644 index 0000000000..1f9ba23a2d --- /dev/null +++ b/apps/docs/mcp-client-channels.mdx @@ -0,0 +1,142 @@ +--- +title: Persistent MCP client channels +description: "How an authenticated MCP client can expose its own tools, resources, and prompts through an Executor session." +--- + +An MCP client channel lets an authenticated client expose a declared set of its +own tools, resources, and prompts through Executor. Executor publishes that +manifest in the MCP session and routes calls back to the client that registered +it. The channel is separate from Executor's ordinary integration catalog and +does not grant access to another client's capabilities. + +This is an Executor-specific protocol, not an MCP transport extension. The +client must authenticate with its normal Executor credentials for registration, +the WebSocket, and MCP requests. + +## Manifest + +The manifest is JSON with three arrays. Tool and prompt entries use `name`; +resource entries use `uri`. + +```json +{ + "tools": [ + { + "name": "read_workspace", + "description": "Read a file from the approved workspace", + "inputSchema": { + "type": "object", + "properties": { "path": { "type": "string" } }, + "required": ["path"] + } + } + ], + "resources": [ + { + "uri": "workspace://README.md", + "name": "README", + "mimeType": "text/markdown" + } + ], + "prompts": [ + { + "name": "review_file", + "description": "Review a workspace file", + "arguments": [{ "name": "path", "required": true }] + } + ] +} +``` + +Only names and descriptions are declarations; they do not enforce permissions +inside the client. The client remains responsible for applying its own access +checks when handling a request. + +## Cloud registration + +Register with `POST /mcp/client-channel` using the usual bearer credentials and +a JSON body containing `manifest`: + +```http +POST /mcp/client-channel +Authorization: Bearer +Content-Type: application/json + +{"manifest":{"tools":[],"resources":[],"prompts":[]}} +``` + +A successful registration returns `201 Created` with a channel token, WebSocket +URL, and initial manifest generation: + +```json +{ + "token": "", + "websocket": "https://executor.example/mcp/client-channel", + "generation": 0 +} +``` + +Connect to the returned URL with a WebSocket upgrade, the same Executor bearer, +and `x-executor-client-token: `. The bearer identifies the +authenticated account and organization; the server-issued token selects that +client's registered manifest. Both checks must succeed before the socket is +accepted. + +## Self-host registration + +Open a WebSocket upgrade to `GET /api/mcp/client-channel` using the normal +authenticated credentials. After the upgrade, send the manifest in the first +frame: + +```json +{ + "type": "initialize", + "manifest": { "tools": [], "resources": [], "prompts": [] } +} +``` + +The server responds with a `bound` frame containing an opaque +`bindingToken`, a server-assigned `clientIdentity`, and `manifestGeneration`. +Keep the token secret. Use it as `x-executor-client-token` on subsequent MCP +HTTP requests, together with the normal Executor credentials. + +## Invocation and replies + +Executor sends capability calls to the registered WebSocket as JSON frames: + +```json +{ + "id": "", + "method": "tools/call", + "params": { "name": "read_workspace", "arguments": { "path": "README.md" } } +} +``` + +Supported methods are `tools/call`, `resources/read`, and `prompts/get`. The +client replies with the same `id` and either a `result` or an `error`: + +```json +{ "id": "", "result": { "content": [{ "type": "text", "text": "..." }] } } +``` + +Executor checks each request against the registered manifest and routes it only +to that channel. MCP discovery lists only the manifest's declared client-owned +capabilities; normal Executor tools remain available according to the session's +usual configuration. + +## Lifecycle and security + +- Treat the channel token as a bearer secret. Send it only in the + `x-executor-client-token` header over HTTPS/WSS; never put it in a URL, + logs, or client-controlled identity fields. +- Keep the WebSocket open while the client expects to receive calls. Closing + the socket invalidates that binding; register and bind a new channel after + reconnecting. +- Use a distinct registration per client instance. The server binds each + registration and MCP session to the authenticated principal, client identity, + and manifest generation. A different client cannot resume that session. +- A manifest describes available capabilities but does not grant authority. + The client must enforce its own tool/resource access policy, and MCP calls + remain subject to Executor's normal authentication and session checks. +- WebSocket messages and manifests are size-limited. Keep manifests concise and + reject requests the client does not recognize. diff --git a/apps/host-selfhost/src/app.ts b/apps/host-selfhost/src/app.ts index 18bcdf9fc2..edc273edb7 100644 --- a/apps/host-selfhost/src/app.ts +++ b/apps/host-selfhost/src/app.ts @@ -26,6 +26,8 @@ import { SelfHostPluginsProvider, } from "./execution"; import { makeSelfHostMcpSeams } from "./mcp"; +import { makeMcpClientChannelRoute } from "./mcp/client-channel"; +import { SelfHostMcpClientBridge } from "./mcp/auth"; import { selfHostPlugins } from "./plugins"; import { ErrorCaptureLive } from "./observability"; import { oauthCallbackSignInRedirectLocation } from "./auth/oauth-callback-login"; @@ -115,6 +117,13 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { }, extensions: { routes: [ + // Persistent in-process MCP client link. The WebSocket upgrade uses + // Effect's request upgrade capability and shares the auth/bridge + // services with ordinary MCP HTTP requests. + makeMcpClientChannelRoute([ + new URL(config.webBaseUrl).origin, + ...config.trustedOrigins.map((origin) => new URL(origin).origin), + ]).pipe(Layer.provideMerge(mcp.auth)), // CLI device-login discovery, must precede the /api/auth/* wildcard // below (Better Auth would otherwise 404 it). The verification page it // points at (/device) is a console SPA route @@ -153,6 +162,7 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { // layer, so operations it serves file as `via: "ui"`). boot: Layer.mergeAll( Layer.succeed(SelfHostDb)(dbHandle), + Layer.succeed(SelfHostMcpClientBridge)(mcp.clientBridge), identityLayer, memberDirectoryLayer, Layer.succeed(ArtifactUsageObserver)((action) => diff --git a/apps/host-selfhost/src/mcp/auth.ts b/apps/host-selfhost/src/mcp/auth.ts index 967d4255d3..af43741c4a 100644 --- a/apps/host-selfhost/src/mcp/auth.ts +++ b/apps/host-selfhost/src/mcp/auth.ts @@ -1,4 +1,4 @@ -import { Effect, Layer } from "effect"; +import { Context, Effect, Layer } from "effect"; import { oAuthDiscoveryMetadata, oAuthProtectedResourceMetadata } from "better-auth/plugins"; import { IdentityProvider, isPlatformPrincipal } from "@executor-js/api/server"; @@ -10,6 +10,7 @@ import { type McpDiscoveryRoute, type Principal, } from "@executor-js/host-mcp"; +import { makeInMemoryMcpClientBridge } from "@executor-js/host-mcp/in-memory-client-bridge"; import { isPrivileged } from "../admin/require-admin"; import { BetterAuth } from "../auth/better-auth"; @@ -59,6 +60,25 @@ const TOOLKIT_PROTECTED_RESOURCE_METADATA_PATH = `${PROTECTED_RESOURCE_METADATA_ const AUTHORIZATION_SERVER_METADATA_PATH = "/.well-known/oauth-authorization-server"; const TOOLKIT_MCP_SEGMENT = "/mcp/toolkits/"; +export class SelfHostMcpClientBridge extends Context.Service< + SelfHostMcpClientBridge, + ReturnType +>()("@executor-js/host-selfhost/SelfHostMcpClientBridge") {} + +export const authenticateClientBinding = ( + principal: Principal, + request: Request, + clientBridge: ReturnType, + challenge: string, +): AuthOutcome => { + const token = request.headers.get("x-executor-client-token"); + if (token === null) return authenticated(principal); + const binding = clientBridge.getBinding(token, principal); + return binding + ? authenticated(principal, binding.clientIdentity, binding.manifestGeneration) + : unauthorized(challenge); +}; + const parseRoles = (role: string | null | undefined): ReadonlyArray => (role ?? "user") .split(",") @@ -153,145 +173,150 @@ const toolkitProtectedResourceMetadata = ( }); }; -export const selfHostMcpAuth: Layer.Layer = - Layer.effect( - McpAuthProvider, - Effect.gen(function* () { - const { auth, organizationId, organizationName, organizationSlug } = yield* BetterAuth; - const fallback = yield* IdentityProvider; +export const selfHostMcpAuth: Layer.Layer< + McpAuthProvider, + never, + BetterAuth | IdentityProvider | SelfHostMcpClientBridge +> = Layer.effect( + McpAuthProvider, + Effect.gen(function* () { + const { auth, organizationId, organizationName, organizationSlug } = yield* BetterAuth; + const fallback = yield* IdentityProvider; + const clientBridge = yield* SelfHostMcpClientBridge; - const asMetadata = oAuthDiscoveryMetadata(auth); - const prMetadata = oAuthProtectedResourceMetadata(auth); + const asMetadata = oAuthDiscoveryMetadata(auth); + const prMetadata = oAuthProtectedResourceMetadata(auth); - const baseURL = auth.options.baseURL; - const resourceMetadataUrl = (request: Request): string => - resourceMetadataUrlFor(baseURL, request); + const baseURL = auth.options.baseURL; + const resourceMetadataUrl = (request: Request): string => + resourceMetadataUrlFor(baseURL, request); - // RFC 9728 challenge string carried on the Unauthorized outcome. Same shape - // as the envelope's default; we supply it explicitly to keep the 401's - // `WWW-Authenticate` fully owned by the provider. - const challengeFor = (request: Request): string => - `Bearer resource_metadata="${resourceMetadataUrl(request)}"`; + // RFC 9728 challenge string carried on the Unauthorized outcome. Same shape + // as the envelope's default; we supply it explicitly to keep the 401's + // `WWW-Authenticate` fully owned by the provider. + const challengeFor = (request: Request): string => + `Bearer resource_metadata="${resourceMetadataUrl(request)}"`; - const discoveryRoutes: ReadonlyArray = [ - { - path: PROTECTED_RESOURCE_METADATA_PATH, - handler: (request) => - Effect.promise(() => prMetadata(request)).pipe( - Effect.flatMap((response) => - toolkitProtectedResourceMetadata(request, response, baseURL), - ), + const discoveryRoutes: ReadonlyArray = [ + { + path: PROTECTED_RESOURCE_METADATA_PATH, + handler: (request) => + Effect.promise(() => prMetadata(request)).pipe( + Effect.flatMap((response) => + toolkitProtectedResourceMetadata(request, response, baseURL), ), - }, - { - path: TOOLKIT_PROTECTED_RESOURCE_METADATA_PATH, - handler: (request) => - Effect.promise(() => prMetadata(request)).pipe( - Effect.flatMap((response) => - toolkitProtectedResourceMetadata(request, response, baseURL), - ), + ), + }, + { + path: TOOLKIT_PROTECTED_RESOURCE_METADATA_PATH, + handler: (request) => + Effect.promise(() => prMetadata(request)).pipe( + Effect.flatMap((response) => + toolkitProtectedResourceMetadata(request, response, baseURL), ), - }, - { - path: AUTHORIZATION_SERVER_METADATA_PATH, - handler: (request) => Effect.promise(() => asMetadata(request)), - }, - ]; - - // Resolved once; `internalAdapter.findUserById` enriches an OAuth userId. - const context = yield* Effect.promise(() => auth.$context); - - /** Enrich a bare OAuth `userId` into the full provider-neutral principal. */ - const principalFromUserId = (userId: string): Effect.Effect => - Effect.gen(function* () { - const user = yield* Effect.promise(() => context.internalAdapter.findUserById(userId)); - if (!user) return null; - // The workspace role, read from the INSTANCE org's membership row - // (an OAuth token carries no session, so the header-based - // `getActiveMemberRole` gate is out of reach — the adapter query - // answers the same question against the same table). FAIL CLOSED to - // "member": an infra fault demotes rather than escalates. - const membership = yield* Effect.promise(() => - context.adapter.findOne<{ readonly role?: string | null }>({ - model: "member", - where: [ - { field: "userId", value: userId }, - { field: "organizationId", value: organizationId }, - ], - }), - ).pipe(Effect.orElseSucceed(() => null)); - const orgRole = - membership?.role != null && isPrivileged(membership.role) - ? ("admin" as const) - : ("member" as const); - return { - accountId: user.id, - // Single-org self-host: OAuth tokens carry no active org, so pin to - // the seeded org (same default as the cookie/api-key path). - organizationId, - organizationName, - organizationSlug, - email: user.email ?? "", - name: user.name ?? null, - avatarUrl: user.image ?? null, - roles: parseRoles(userRole(user)), - orgRoleModel: "organization", - orgRole, - } satisfies Principal; - }); + ), + }, + { + path: AUTHORIZATION_SERVER_METADATA_PATH, + handler: (request) => Effect.promise(() => asMetadata(request)), + }, + ]; - /** (a) The mcp() OAuth opaque bearer, with self-enforced expiry. */ - const authenticateOAuthBearer = (request: Request): Effect.Effect => - Effect.gen(function* () { - const session = yield* Effect.promise(() => - auth.api.getMcpSession({ headers: request.headers }), - ); - if (!session) return null; - // GOTCHA: getMcpSession does NOT validate accessTokenExpiresAt — an - // expired token still resolves. Reject it here. - if (new Date(session.accessTokenExpiresAt).getTime() < Date.now()) return null; - return yield* principalFromUserId(session.userId); - }).pipe(Effect.orElseSucceed(() => null)); + // Resolved once; `internalAdapter.findUserById` enriches an OAuth userId. + const context = yield* Effect.promise(() => auth.$context); - /** (b) The existing cookie / bearer-session / x-api-key path. The fallback's - * api `Principal` shape is byte-identical to host-mcp's `Principal`. The - * neutral seam can also resolve a platform credential, which self-host's - * identity never produces — narrowed away rather than asserted, so an MCP - * session can never bind to a subject-less credential if that changes. */ - const authenticateSession = (request: Request): Effect.Effect => - fallback.authenticate(request).pipe( - Effect.map((principal) => (isPlatformPrincipal(principal) ? null : principal)), - Effect.catchTags({ - Unauthorized: () => Effect.succeed(null), - NoOrganization: () => Effect.succeed(null), - ReadOnlyCredential: () => Effect.succeed(null), + /** Enrich a bare OAuth `userId` into the full provider-neutral principal. */ + const principalFromUserId = (userId: string): Effect.Effect => + Effect.gen(function* () { + const user = yield* Effect.promise(() => context.internalAdapter.findUserById(userId)); + if (!user) return null; + // The workspace role, read from the INSTANCE org's membership row + // (an OAuth token carries no session, so the header-based + // `getActiveMemberRole` gate is out of reach — the adapter query + // answers the same question against the same table). FAIL CLOSED to + // "member": an infra fault demotes rather than escalates. + const membership = yield* Effect.promise(() => + context.adapter.findOne<{ readonly role?: string | null }>({ + model: "member", + where: [ + { field: "userId", value: userId }, + { field: "organizationId", value: organizationId }, + ], }), - ); + ).pipe(Effect.orElseSucceed(() => null)); + const orgRole = + membership?.role != null && isPrivileged(membership.role) + ? ("admin" as const) + : ("member" as const); + return { + accountId: user.id, + // Single-org self-host: OAuth tokens carry no active org, so pin to + // the seeded org (same default as the cookie/api-key path). + organizationId, + organizationName, + organizationSlug, + email: user.email ?? "", + name: user.name ?? null, + avatarUrl: user.image ?? null, + roles: parseRoles(userRole(user)), + orgRoleModel: "organization", + orgRole, + } satisfies Principal; + }); - /** - * Try the OAuth bearer ONLY when a Bearer header is present (no - * getMcpSession round-trip on cookie requests), then the cookie/api-key - * fallback. Self-host always pins an org, so the outcome is always - * Authenticated or Unauthorized. - */ - const authenticate = (request: Request): Effect.Effect => - (hasBearer(request) - ? authenticateOAuthBearer(request).pipe( - Effect.flatMap((principal) => - principal ? Effect.succeed(principal) : authenticateSession(request), - ), - ) - : authenticateSession(request) - ).pipe( - Effect.map((principal) => - principal ? authenticated(principal) : unauthorized(challengeFor(request)), - ), + /** (a) The mcp() OAuth opaque bearer, with self-enforced expiry. */ + const authenticateOAuthBearer = (request: Request): Effect.Effect => + Effect.gen(function* () { + const session = yield* Effect.promise(() => + auth.api.getMcpSession({ headers: request.headers }), ); + if (!session) return null; + // GOTCHA: getMcpSession does NOT validate accessTokenExpiresAt — an + // expired token still resolves. Reject it here. + if (new Date(session.accessTokenExpiresAt).getTime() < Date.now()) return null; + return yield* principalFromUserId(session.userId); + }).pipe(Effect.orElseSucceed(() => null)); + + /** (b) The existing cookie / bearer-session / x-api-key path. The fallback's + * api `Principal` shape is byte-identical to host-mcp's `Principal`. The + * neutral seam can also resolve a platform credential, which self-host's + * identity never produces — narrowed away rather than asserted, so an MCP + * session can never bind to a subject-less credential if that changes. */ + const authenticateSession = (request: Request): Effect.Effect => + fallback.authenticate(request).pipe( + Effect.map((principal) => (isPlatformPrincipal(principal) ? null : principal)), + Effect.catchTags({ + Unauthorized: () => Effect.succeed(null), + NoOrganization: () => Effect.succeed(null), + ReadOnlyCredential: () => Effect.succeed(null), + }), + ); + + /** + * Try the OAuth bearer ONLY when a Bearer header is present (no + * getMcpSession round-trip on cookie requests), then the cookie/api-key + * fallback. Self-host always pins an org, so the outcome is always + * Authenticated or Unauthorized. + */ + const authenticate = (request: Request): Effect.Effect => + (hasBearer(request) + ? authenticateOAuthBearer(request).pipe( + Effect.flatMap((principal) => + principal ? Effect.succeed(principal) : authenticateSession(request), + ), + ) + : authenticateSession(request) + ).pipe( + Effect.map((principal) => { + if (!principal) return unauthorized(challengeFor(request)); + return authenticateClientBinding(principal, request, clientBridge, challengeFor(request)); + }), + ); - return { - discoveryRoutes, - resourceMetadataUrl, - authenticate, - }; - }), - ); + return { + discoveryRoutes, + resourceMetadataUrl, + authenticate, + }; + }), +); diff --git a/apps/host-selfhost/src/mcp/client-binding.test.ts b/apps/host-selfhost/src/mcp/client-binding.test.ts new file mode 100644 index 0000000000..f59c78aabe --- /dev/null +++ b/apps/host-selfhost/src/mcp/client-binding.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { makeInMemoryMcpClientBridge } from "@executor-js/host-mcp/in-memory-client-bridge"; +import type { Principal } from "@executor-js/host-mcp"; +import { authenticateClientBinding } from "./auth"; + +const owner: Principal = { + accountId: "account", + organizationId: "organization", + organizationName: "Organization", + email: "owner@example.test", + name: "Owner", + avatarUrl: null, + roles: [], + orgRoleModel: "none", +}; +const manifest = { + tools: [{ name: "hello" }], + resources: [], + prompts: [], +}; + +describe("self-host MCP client binding authentication", () => { + it("keeps requests remote-only when no client binding token is supplied", () => { + const bridge = makeInMemoryMcpClientBridge(); + const outcome = authenticateClientBinding( + owner, + new Request("http://localhost/mcp"), + bridge, + "Bearer challenge", + ); + expect(outcome).toEqual({ _tag: "Authenticated", principal: owner }); + }); + + it("rejects invalid or cross-principal tokens and ignores spoofed identity", () => { + const bridge = makeInMemoryMcpClientBridge(); + const connection = bridge.connect({ principal: owner, manifest, send: () => {} }); + const spoofed = new Headers({ + "x-executor-client-token": connection.bindingToken, + "x-executor-client-identity": "attacker-selected", + }); + const bound = authenticateClientBinding( + owner, + new Request("http://localhost/mcp", { headers: spoofed }), + bridge, + "Bearer challenge", + ); + expect(bound).toEqual({ + _tag: "Authenticated", + principal: owner, + clientIdentity: bridge.getBinding(connection.bindingToken, owner)?.clientIdentity, + clientManifestGeneration: connection.manifestGeneration, + }); + const outcome = authenticateClientBinding( + { ...owner, accountId: "other-account" }, + new Request("http://localhost/mcp", { headers: spoofed }), + bridge, + "Bearer challenge", + ); + expect(outcome).toEqual({ _tag: "Unauthorized", challenge: "Bearer challenge" }); + expect( + authenticateClientBinding( + owner, + new Request("http://localhost/mcp", { + headers: { "x-executor-client-token": "not-a-binding" }, + }), + bridge, + "Bearer challenge", + ), + ).toEqual({ _tag: "Unauthorized", challenge: "Bearer challenge" }); + }); + + it("creates a fresh identity and generation for every reconnect", () => { + const bridge = makeInMemoryMcpClientBridge(); + const first = bridge.connect({ principal: owner, manifest, send: () => {} }); + const second = bridge.connect({ principal: owner, manifest, send: () => {} }); + const firstBinding = bridge.getBinding(first.bindingToken, owner); + const secondBinding = bridge.getBinding(second.bindingToken, owner); + + expect(secondBinding?.clientIdentity).not.toBe(firstBinding?.clientIdentity); + expect(secondBinding?.manifestGeneration).toBeGreaterThan( + firstBinding?.manifestGeneration ?? 0, + ); + expect(bridge.authenticate(first.bindingToken, owner)).toBe(firstBinding?.clientIdentity); + expect(bridge.authenticate(second.bindingToken, owner)).toBe(secondBinding?.clientIdentity); + }); +}); diff --git a/apps/host-selfhost/src/mcp/client-channel-manifest.test.ts b/apps/host-selfhost/src/mcp/client-channel-manifest.test.ts new file mode 100644 index 0000000000..5bc259dd62 --- /dev/null +++ b/apps/host-selfhost/src/mcp/client-channel-manifest.test.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + isClientChannelFrameWithinLimit, + isValidMcpClientManifest, + MCP_CLIENT_CHANNEL_MAX_FRAME_LENGTH, +} from "./client-channel"; + +describe("self-host MCP client channel manifest validation", () => { + it("accepts a valid capability manifest", () => { + expect( + isValidMcpClientManifest({ + tools: [{ name: "read" }], + resources: [{ uri: "workspace://README.md" }], + prompts: [{ name: "review" }], + }), + ).toBe(true); + }); + + it("rejects manifests over the capability count and duplicate-name limits", () => { + expect( + isValidMcpClientManifest({ + tools: Array.from({ length: 101 }, (_, index) => ({ name: `tool-${index}` })), + resources: [], + prompts: [], + }), + ).toBe(false); + expect( + isValidMcpClientManifest({ + tools: [{ name: "same" }, { name: "same" }], + resources: [], + prompts: [], + }), + ).toBe(false); + }); + + it("rejects invalid and oversized channel frames", () => { + expect( + isValidMcpClientManifest({ + tools: [{ name: "" }], + resources: [], + prompts: [], + }), + ).toBe(false); + expect(isClientChannelFrameWithinLimit("x".repeat(MCP_CLIENT_CHANNEL_MAX_FRAME_LENGTH))).toBe( + true, + ); + expect( + isClientChannelFrameWithinLimit("x".repeat(MCP_CLIENT_CHANNEL_MAX_FRAME_LENGTH + 1)), + ).toBe(false); + }); +}); diff --git a/apps/host-selfhost/src/mcp/client-channel-origin.test.ts b/apps/host-selfhost/src/mcp/client-channel-origin.test.ts new file mode 100644 index 0000000000..6d464bdc5b --- /dev/null +++ b/apps/host-selfhost/src/mcp/client-channel-origin.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { hasAllowedCookieOrigin } from "./client-channel"; + +describe("MCP client channel cookie origin guard", () => { + const allowedOrigins = ["https://executor.example", "https://console.example"]; + + it("requires an allowed Origin whenever Cookie is present", () => { + expect( + hasAllowedCookieOrigin( + new Request("https://executor.example/api/mcp/client-channel", { + headers: { cookie: "session=abc", origin: "https://attacker.example" }, + }), + allowedOrigins, + ), + ).toBe(false); + expect( + hasAllowedCookieOrigin( + new Request("https://executor.example/api/mcp/client-channel", { + headers: { cookie: "session=abc" }, + }), + allowedOrigins, + ), + ).toBe(false); + expect( + hasAllowedCookieOrigin( + new Request("https://executor.example/api/mcp/client-channel", { + headers: { cookie: "session=abc", origin: "https://console.example" }, + }), + allowedOrigins, + ), + ).toBe(true); + }); + + it("allows bearer/API-key clients without Cookie and without Origin", () => { + expect( + hasAllowedCookieOrigin( + new Request("https://executor.example/api/mcp/client-channel", { + headers: { authorization: "Bearer token" }, + }), + allowedOrigins, + ), + ).toBe(true); + }); +}); diff --git a/apps/host-selfhost/src/mcp/client-channel.ts b/apps/host-selfhost/src/mcp/client-channel.ts new file mode 100644 index 0000000000..a2dc09c501 --- /dev/null +++ b/apps/host-selfhost/src/mcp/client-channel.ts @@ -0,0 +1,176 @@ +import { Effect, Predicate, Schema } from "effect"; +import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; + +import type { McpClientFrame } from "@executor-js/host-mcp/in-memory-client-bridge"; +import { McpAuthProvider } from "@executor-js/host-mcp"; +import { SelfHostMcpClientBridge } from "./auth"; + +const McpClientManifestSchema = Schema.Struct({ + tools: Schema.Array( + Schema.Struct({ + name: Schema.String, + description: Schema.optional(Schema.String), + inputSchema: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), + }), + ), + resources: Schema.Array( + Schema.Struct({ + uri: Schema.String, + name: Schema.optional(Schema.String), + description: Schema.optional(Schema.String), + mimeType: Schema.optional(Schema.String), + }), + ), + prompts: Schema.Array( + Schema.Struct({ + name: Schema.String, + description: Schema.optional(Schema.String), + arguments: Schema.optional( + Schema.Array( + Schema.Struct({ + name: Schema.String, + description: Schema.optional(Schema.String), + required: Schema.optional(Schema.Boolean), + }), + ), + ), + }), + ), +}); +const hasManifestShape = Schema.is(McpClientManifestSchema); + +export const MCP_CLIENT_CHANNEL_MAX_FRAME_LENGTH = 65_536; + +export const isClientChannelFrameWithinLimit = (raw: string): boolean => + raw.length <= MCP_CLIENT_CHANNEL_MAX_FRAME_LENGTH; + +export const isValidMcpClientManifest = ( + value: unknown, +): value is Schema.Schema.Type => { + if (!hasManifestShape(value)) return false; + const nameGroups = [ + value.tools.map((item) => item.name), + value.resources.map((item) => item.uri), + value.prompts.map((item) => item.name), + ]; + return ( + value.tools.length + value.resources.length + value.prompts.length <= 100 && + nameGroups.every( + (names) => + names.every((name) => name.length > 0 && name.length <= 200) && + new Set(names).size === names.length, + ) + ); +}; + +const response = (status: number, message: string, headers?: HeadersInit) => + HttpServerResponse.text(message, { status, headers }); + +export const hasAllowedCookieOrigin = ( + request: Request, + allowedOrigins: ReadonlyArray, +): boolean => { + if (!request.headers.has("cookie")) return true; + const origin = request.headers.get("origin"); + return origin !== null && allowedOrigins.includes(origin); +}; + +const routeFor = (allowedOrigins: ReadonlyArray) => + Effect.gen(function* () { + const incoming = yield* HttpServerRequest.HttpServerRequest; + const auth = yield* McpAuthProvider; + const bridge = yield* SelfHostMcpClientBridge; + const webRequest = + incoming.source instanceof Request + ? incoming.source + : new Request( + new URL( + incoming.url, + `http://${(incoming.headers as Record).host ?? "host.internal"}`, + ), + { + method: "GET", + headers: new Headers(incoming.headers as Record), + }, + ); + if (!hasAllowedCookieOrigin(webRequest, allowedOrigins)) { + return response(403, "Forbidden"); + } + const outcome = yield* auth.authenticate(webRequest); + if (!Predicate.isTagged("Authenticated")(outcome)) { + const challenge = Predicate.isTagged("Unauthorized")(outcome) + ? (outcome.challenge ?? + `Bearer resource_metadata="${auth.resourceMetadataUrl(webRequest)}"`) + : undefined; + return response( + 401, + "Unauthorized", + challenge ? { "www-authenticate": challenge } : undefined, + ); + } + + const socket = yield* incoming.upgrade; + const write = yield* socket.writer; + const context = yield* Effect.context(); + let connection: ReturnType | null = null; + let initialized = false; + + yield* socket + .runString((raw) => + Effect.gen(function* () { + if (!isClientChannelFrameWithinLimit(raw)) { + yield* write(JSON.stringify({ type: "error", message: "Client frame too large" })); + return; + } + const message = yield* Schema.decodeUnknownEffect(Schema.UnknownFromJsonString)(raw).pipe( + Effect.orElseSucceed(() => null), + ); + if (typeof message !== "object" || message === null) return; + const frame = message as Record; + + if (!initialized) { + if (frame.type !== "initialize" || !isValidMcpClientManifest(frame.manifest)) { + yield* write(JSON.stringify({ type: "error", message: "Invalid initial manifest" })); + return; + } + initialized = true; + connection = bridge.connect({ + principal: outcome.principal, + manifest: frame.manifest, + send: (outbound: McpClientFrame) => { + // The bridge transport contract is synchronous; schedule writes + // on the acquired Effect socket writer. + void Effect.runPromiseWith(context)(write(JSON.stringify(outbound))); + }, + }); + const binding = bridge.getBinding(connection.bindingToken, outcome.principal); + if (binding === null) { + connection.disconnect(); + connection = null; + yield* write(JSON.stringify({ type: "error", message: "Unable to create binding" })); + return; + } + yield* write( + JSON.stringify({ + type: "bound", + bindingToken: connection.bindingToken, + clientIdentity: binding.clientIdentity, + manifestGeneration: connection.manifestGeneration, + }), + ); + return; + } + + if (!connection || typeof frame.id !== "string") return; + if (Object.hasOwn(frame, "error")) connection.respond(frame.id, undefined, frame.error); + else if (Object.hasOwn(frame, "result")) connection.respond(frame.id, frame.result); + }), + ) + .pipe(Effect.ensuring(Effect.sync(() => connection?.disconnect()))); + + return HttpServerResponse.empty(); + }); + +/** Effect HTTP route for the authenticated, persistent client link. */ +export const makeMcpClientChannelRoute = (allowedOrigins: ReadonlyArray) => + HttpRouter.add("GET", "/api/mcp/client-channel", routeFor(allowedOrigins)); diff --git a/apps/host-selfhost/src/mcp/index.ts b/apps/host-selfhost/src/mcp/index.ts index c8a8683227..6d130daa95 100644 --- a/apps/host-selfhost/src/mcp/index.ts +++ b/apps/host-selfhost/src/mcp/index.ts @@ -12,7 +12,8 @@ import { BetterAuth, type BetterAuthHandle } from "../auth/better-auth"; import { resolveSelfHostOrgRole } from "../auth/identity"; import type { SelfHostDbHandle } from "../db/self-host-db"; import type { SelfHostConfig } from "../config"; -import { selfHostMcpAuth } from "./auth"; +import { makeInMemoryMcpClientBridge } from "@executor-js/host-mcp/in-memory-client-bridge"; +import { selfHostMcpAuth, SelfHostMcpClientBridge } from "./auth"; import { makeSelfHostMcpSessionStore, selfHostMcpReporter, @@ -64,6 +65,8 @@ export interface SelfHostMcpSeams { * `/resume`) for the console approval page. Browser elicitation mode only. */ readonly approvalHandler: (request: Request) => Promise; + /** Ephemeral client bindings used by the persistent client channel. */ + readonly clientBridge: ReturnType; /** Dispose all live in-process MCP sessions at shutdown (not a seam). */ readonly close: () => Promise; } @@ -154,21 +157,25 @@ export const makeSelfHostMcpSeams = ( betterAuth: BetterAuthHandle, config: SelfHostConfig, ): SelfHostMcpSeams => { + const clientBridge = makeInMemoryMcpClientBridge(); // The pinned public origin keeps browser-approval URLs reachable behind a // reverse proxy (not the internal 127.0.0.1 bind from the request URL). const sessionStore = makeSelfHostMcpSessionStore( dbHandle, config.webBaseUrl, config.mcpSessionIdleTtlMs, + clientBridge, ); const auth: Layer.Layer = selfHostMcpAuth.pipe( Layer.provide(Layer.succeed(BetterAuth)(betterAuth)), + Layer.provide(Layer.succeed(SelfHostMcpClientBridge)(clientBridge)), ); return { auth, sessions: selfHostMcpSessions(sessionStore), reporter: selfHostMcpReporter, approvalHandler: makeApprovalHandler(sessionStore, betterAuth), + clientBridge, close: sessionStore.close, }; }; diff --git a/apps/host-selfhost/src/mcp/mcp-oauth.test.ts b/apps/host-selfhost/src/mcp/mcp-oauth.test.ts index a66952a0e1..4b45c95c20 100644 --- a/apps/host-selfhost/src/mcp/mcp-oauth.test.ts +++ b/apps/host-selfhost/src/mcp/mcp-oauth.test.ts @@ -63,6 +63,16 @@ test("an unauthenticated /mcp request returns 401 with a WWW-Authenticate challe expect(challenge).toContain("resource_metadata="); }); +test("the persistent client channel rejects an unauthenticated upgrade", async () => { + const response = await handler( + new Request(`${BASE}/api/mcp/client-channel`, { + headers: { upgrade: "websocket", connection: "Upgrade" }, + }), + ); + expect(response.status).toBe(401); + expect(response.headers.get("www-authenticate")).toContain("Bearer"); +}); + test("an unauthenticated toolkit MCP request challenges with toolkit metadata", async () => { const res = await handler( new Request(`${BASE}/mcp/toolkits/deploy`, { diff --git a/apps/host-selfhost/src/mcp/session-store.ts b/apps/host-selfhost/src/mcp/session-store.ts index f9d29b1f05..a1d574ee04 100644 --- a/apps/host-selfhost/src/mcp/session-store.ts +++ b/apps/host-selfhost/src/mcp/session-store.ts @@ -7,6 +7,7 @@ import { makeInMemoryMcpSessionStore, type InMemoryMcpSessionStore, } from "@executor-js/host-mcp/in-memory-session-store"; +import type { McpClientSessionResolver } from "@executor-js/host-mcp/in-memory-client-bridge"; import { selfHostAnalytics } from "../analytics"; import { ErrorCaptureLive } from "../observability"; @@ -37,6 +38,7 @@ export const makeSelfHostMcpSessionStore = ( db: SelfHostDbHandle, webBaseUrl?: string, sessionIdleTtlMs?: number, + clientSessionResolver?: McpClientSessionResolver, ): InMemoryMcpSessionStore => makeInMemoryMcpSessionStore( makeMcpBuildServer( @@ -52,6 +54,7 @@ export const makeSelfHostMcpSessionStore = ( { ...(webBaseUrl === undefined ? {} : { webBaseUrl }), ...(sessionIdleTtlMs === undefined ? {} : { sessionIdleTtlMs }), + ...(clientSessionResolver === undefined ? {} : { clientSessionResolver }), }, ); diff --git a/e2e/cloud/mcp-client-channel.test.ts b/e2e/cloud/mcp-client-channel.test.ts new file mode 100644 index 0000000000..cdeb7ed5d7 --- /dev/null +++ b/e2e/cloud/mcp-client-channel.test.ts @@ -0,0 +1,326 @@ +import { randomBytes } from "node:crypto"; +import { request as httpRequest } from "node:http"; +import { request as httpsRequest } from "node:https"; +import type { Duplex } from "node:stream"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { Client } from "@modelcontextprotocol/sdk/client/index.js"; +import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; + +import { scenario } from "../src/scenario"; +import { Mcp, Target } from "../src/services"; +import type { Identity } from "../src/target"; + +const emailOf = (identity: Identity): string => identity.credentials?.email ?? identity.label; +const CLIENT_TOOL = "channel_echo"; +const CLIENT_RESOURCE = "client://shared-item"; +const CLIENT_PROMPT = "shared_prompt"; + +interface Registration { + readonly token: string; + readonly websocket: string; +} + +interface ClientFrame { + readonly id: string; + readonly method: string; + readonly params: Record; +} + +/** Minimal masked RFC 6455 client used because Node's built-in WebSocket + * cannot set the bearer and opaque binding headers required by this route. */ +class HeaderWebSocket { + private buffer = Buffer.alloc(0); + private onFrame: ((frame: ClientFrame) => void) | undefined; + + private constructor( + private readonly socket: Duplex, + initial: Buffer, + ) { + socket.on("data", (chunk: Buffer) => this.consume(chunk)); + socket.on("close", () => undefined); + socket.on("error", () => undefined); + if (initial.length > 0) this.consume(initial); + } + + static connect(url: string, bearer: string, token: string): Promise { + const target = new URL(url); + return Effect.runPromise( + Effect.callback((resume) => { + const makeRequest = target.protocol === "https:" ? httpsRequest : httpRequest; + const request = makeRequest({ + protocol: target.protocol, + hostname: target.hostname, + port: target.port, + path: `${target.pathname}${target.search}`, + method: "GET", + headers: { + authorization: `Bearer ${bearer}`, + connection: "Upgrade", + upgrade: "websocket", + "sec-websocket-version": "13", + "sec-websocket-key": randomBytes(16).toString("base64"), + "x-executor-client-token": token, + }, + }); + const timeout = setTimeout( + () => request.destroy(new Error("WebSocket upgrade timed out")), + 15_000, + ); + request.once("upgrade", (_response, socket, head) => { + clearTimeout(timeout); + resume(Effect.succeed(new HeaderWebSocket(socket, head))); + }); + request.once("response", (response) => { + clearTimeout(timeout); + response.resume(); + resume(Effect.fail(new Error(`WebSocket upgrade rejected: ${response.statusCode}`))); + }); + request.once("error", (error) => { + clearTimeout(timeout); + resume(Effect.fail(error)); + }); + request.end(); + return Effect.sync(() => request.destroy()); + }), + ); + } + + onRequest(handler: (frame: ClientFrame) => void): void { + this.onFrame = handler; + } + + send(value: unknown): void { + const payload = Buffer.from(JSON.stringify(value)); + const mask = randomBytes(4); + let header: Buffer; + if (payload.length < 126) { + header = Buffer.from([0x81, 0x80 | payload.length]); + } else if (payload.length <= 0xffff) { + header = Buffer.alloc(4); + header[0] = 0x81; + header[1] = 0x80 | 126; + header.writeUInt16BE(payload.length, 2); + } else { + header = Buffer.alloc(10); + header[0] = 0x81; + header[1] = 0x80 | 127; + header.writeBigUInt64BE(BigInt(payload.length), 2); + } + const masked = Buffer.from(payload); + for (let index = 0; index < masked.length; index++) { + masked[index] ^= mask[index % mask.length]!; + } + this.socket.write(Buffer.concat([header, mask, masked])); + } + + close(): void { + const mask = randomBytes(4); + this.socket.write(Buffer.from([0x88, 0x80, ...mask])); + this.socket.end(); + } + + private consume(chunk: Buffer): void { + this.buffer = Buffer.concat([this.buffer, chunk]); + while (this.buffer.length >= 2) { + const first = this.buffer[0]!; + const second = this.buffer[1]!; + let length = second & 0x7f; + let offset = 2; + if (length === 126) { + if (this.buffer.length < 4) return; + length = this.buffer.readUInt16BE(2); + offset = 4; + } else if (length === 127) { + if (this.buffer.length < 10) return; + const longLength = this.buffer.readBigUInt64BE(2); + if (longLength > BigInt(Number.MAX_SAFE_INTEGER)) { + this.socket.destroy(new Error("WebSocket frame too large")); + return; + } + length = Number(longLength); + offset = 10; + } + const maskLength = second & 0x80 ? 4 : 0; + if (this.buffer.length < offset + maskLength + length) return; + const mask = maskLength ? this.buffer.subarray(offset, offset + 4) : null; + offset += maskLength; + const payload = Buffer.from(this.buffer.subarray(offset, offset + length)); + this.buffer = this.buffer.subarray(offset + length); + if (mask) { + for (let index = 0; index < payload.length; index++) { + payload[index] ^= mask[index % mask.length]!; + } + } + const opcode = first & 0x0f; + if (opcode === 0x8) { + this.socket.end(); + return; + } + if (opcode === 0x9) { + this.socket.write(Buffer.from([0x8a, payload.length, ...payload])); + continue; + } + if (opcode !== 0x1) continue; + const parsed: unknown = JSON.parse(payload.toString("utf8")); + if ( + typeof parsed === "object" && + parsed !== null && + "id" in parsed && + typeof parsed.id === "string" && + "method" in parsed && + typeof parsed.method === "string" && + "params" in parsed && + typeof parsed.params === "object" && + parsed.params !== null + ) { + this.onFrame?.(parsed as ClientFrame); + } + } + } +} + +const register = async ( + baseUrl: string, + bearer: string, + owner: "A" | "B", +): Promise => { + const response = await fetch(new URL("/mcp/client-channel", baseUrl), { + method: "POST", + headers: { + authorization: `Bearer ${bearer}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + manifest: { + tools: [ + { + name: CLIENT_TOOL, + description: `Client ${owner} tool`, + inputSchema: { + type: "object", + properties: { value: { type: "string" } }, + required: ["value"], + }, + }, + ], + resources: [{ uri: CLIENT_RESOURCE, name: `Client ${owner} resource` }], + prompts: [ + { + name: CLIENT_PROMPT, + description: `Client ${owner} prompt`, + arguments: [{ name: "topic", required: true }], + }, + ], + }, + }), + }); + expect(response.status, "registration is accepted only through cloud bearer auth").toBe(201); + return (await response.json()) as Registration; +}; + +const connectMcp = async (url: string, bearer: string, token: string) => { + const client = new Client({ name: "client-channel-e2e", version: "1.0.0" }, { capabilities: {} }); + const transport = new StreamableHTTPClientTransport(new URL(url), { + requestInit: { + headers: { + authorization: `Bearer ${bearer}`, + "x-executor-client-token": token, + }, + }, + }); + await client.connect(transport); + return client; +}; + +scenario( + "Cloud · client channel binds manifests and correlates calls to each owner", + {}, + Effect.gen(function* () { + const target = yield* Target; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const bearer = yield* mcp.mintBearer(emailOf(identity)); + const sessionUrl = target.mcpUrl; + const [registrationA, registrationB] = yield* Effect.promise(() => + Promise.all([register(target.baseUrl, bearer, "A"), register(target.baseUrl, bearer, "B")]), + ); + expect(registrationA.token).not.toBe(registrationB.token); + expect(new URL(registrationA.websocket).search).toBe(""); + + yield* Effect.promise(async () => { + let channels: HeaderWebSocket[] = []; + let clients: Client[] = []; + try { + channels.push( + await HeaderWebSocket.connect(registrationA.websocket, bearer, registrationA.token), + ); + channels.push( + await HeaderWebSocket.connect(registrationB.websocket, bearer, registrationB.token), + ); + const attachOwner = (socket: HeaderWebSocket, owner: "A" | "B") => { + socket.onRequest((frame) => { + const result = + frame.method === "tools/call" + ? { content: [{ type: "text", text: `owner-${owner}` }] } + : frame.method === "resources/read" + ? { contents: [{ uri: CLIENT_RESOURCE, text: `owner-${owner}` }] } + : { + messages: [ + { + role: "user", + content: { type: "text", text: `owner-${owner}` }, + }, + ], + }; + socket.send({ id: frame.id, result }); + }); + }; + attachOwner(channels[0]!, "A"); + attachOwner(channels[1]!, "B"); + clients.push(await connectMcp(sessionUrl, bearer, registrationA.token)); + clients.push(await connectMcp(sessionUrl, bearer, registrationB.token)); + const [clientA, clientB] = clients as [Client, Client]; + const toolsA = await clientA.listTools(); + const toolsB = await clientB.listTools(); + expect(toolsA.tools.find((tool) => tool.name === CLIENT_TOOL)?.description).toBe( + "Client A tool", + ); + expect(toolsB.tools.find((tool) => tool.name === CLIENT_TOOL)?.description).toBe( + "Client B tool", + ); + + for (const [client, owner] of [ + [clientA, "A"], + [clientB, "B"], + ] as const) { + const tool = await client.callTool({ + name: CLIENT_TOOL, + arguments: { value: owner }, + }); + expect( + ((tool.content as Array<{ text?: string }>)[0] as { text?: string }).text, + `same-named tool invocation reaches client ${owner}`, + ).toBe(`owner-${owner}`); + const resource = await client.readResource({ uri: CLIENT_RESOURCE }); + expect(resource.contents[0]).toMatchObject({ + uri: CLIENT_RESOURCE, + text: `owner-${owner}`, + }); + const prompt = await client.getPrompt({ + name: CLIENT_PROMPT, + arguments: { topic: owner }, + }); + expect(prompt.messages[0]?.content).toMatchObject({ + type: "text", + text: `owner-${owner}`, + }); + } + } finally { + channels.forEach((channel) => channel.close()); + await Promise.all(clients.map((client) => client.close())); + } + }); + }), +); diff --git a/packages/core/api/src/server/mcp-build.ts b/packages/core/api/src/server/mcp-build.ts index 3891f6cae3..3c75ea3441 100644 --- a/packages/core/api/src/server/mcp-build.ts +++ b/packages/core/api/src/server/mcp-build.ts @@ -50,6 +50,12 @@ export const makeMcpBuildServer = principal.organizationName, { mcpResource: options?.resource, + ...(options?.clientIdentity === undefined + ? {} + : { clientIdentity: options.clientIdentity }), + ...(options?.clientManifestGeneration === undefined + ? {} + : { clientManifestGeneration: options.clientManifestGeneration }), orgWrites: "request", }, ).pipe(Effect.withSpan("mcp.execution_stack.build")); diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index 749839be3e..b9a88edde0 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -218,6 +218,10 @@ export const buildOAuthRedirectUri = (input: { export interface PluginsProviderContext { readonly mcpResource?: McpResource; + /** Server-verified client installation identity, if the session is linked. */ + readonly clientIdentity?: string; + /** Capability manifest revision pinned to this MCP session. */ + readonly clientManifestGeneration?: number; } export interface PluginsProviderShape { diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts index 813f4d5c40..95efed3b68 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.test.ts @@ -14,6 +14,7 @@ import { McpAgentSessionDOBase, type McpApprovalOwner, type McpApprovalPrincipal, + type McpSessionOwnerIdentity, type McpSessionResumeApprovalResult, type McpSessionModelResumeResult, type SessionMeta, @@ -196,10 +197,7 @@ type HarnessSession = { readonly orgWriteAccess: "allowed" | "denied"; } | null>; validateMcpSessionOwner: ( - identity: { - readonly accountId: string; - readonly organizationId: string; - }, + identity: McpSessionOwnerIdentity, resource: McpResource, ) => Promise<"ok" | "not_found" | "forbidden" | "terminated">; }; @@ -682,6 +680,50 @@ describe("McpAgentSessionDOBase transport restore", () => { ).resolves.toBe("forbidden"); }); + it("rejects a different authenticated client for the same owner and resource", async () => { + const session = await makeHarnessSession(); + session.sessionMeta = { + ...session.sessionMeta!, + clientIdentity: "client-a", + clientManifestGeneration: 1, + }; + + await expect( + session.validateMcpSessionOwner( + { accountId: "user-1", organizationId: "org-1", clientIdentity: "client-b" }, + defaultMcpResource, + ), + ).resolves.toBe("forbidden"); + await expect( + session.validateMcpSessionOwner( + { accountId: "user-1", organizationId: "org-1", clientIdentity: "client-a" }, + defaultMcpResource, + ), + ).resolves.toBe("forbidden"); + await expect( + session.validateMcpSessionOwner( + { + accountId: "user-1", + organizationId: "org-1", + clientIdentity: "client-a", + clientManifestGeneration: 1, + }, + defaultMcpResource, + ), + ).resolves.toBe("ok"); + await expect( + session.validateMcpSessionOwner( + { + accountId: "user-1", + organizationId: "org-1", + clientIdentity: "client-a", + clientManifestGeneration: 2, + }, + defaultMcpResource, + ), + ).resolves.toBe("forbidden"); + }); + it("single-flights concurrent same-session restore after idle disposal", async () => { const session = await makeHarnessSession(); const firstRestoreEntered = makeDeferred(); @@ -858,7 +900,10 @@ describe("McpAgentSessionDOBase init survives a platform reset of its bookkeepin buildMcpServer: () => Effect.Effect<{ mcpServer: McpServer; engine: unknown }>; openSessionDb: () => { readonly end: () => void }; resolveSessionMeta: () => Effect.Effect; - validateMcpSessionOwner: (identity: McpApprovalOwner, resource: McpResource) => Promise; + validateMcpSessionOwner: ( + identity: McpSessionOwnerIdentity, + resource: McpResource, + ) => Promise; }; const sessionMeta: SessionMeta = { diff --git a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts index 805dd14d16..e9681c637f 100644 --- a/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts +++ b/packages/hosts/cloudflare/src/mcp/agent-session-durable-object.ts @@ -73,6 +73,10 @@ interface McpSessionInitBase { /** The organization's URL slug, from the same resolved record. */ readonly organizationSlug?: string; readonly userId: string; + /** Verified persistent-client identity, pinned for the session lifetime. */ + readonly clientIdentity?: string; + /** Capability manifest revision, pinned for the session lifetime. */ + readonly clientManifestGeneration?: number; readonly elicitationMode: McpElicitationMode; /** Whether this session serves artifacts, read off `?artifacts=` at connect * time. Absent means the default (enabled). */ @@ -103,6 +107,11 @@ export type McpApprovalOwner = { readonly organizationId: string; }; +export type McpSessionOwnerIdentity = McpApprovalOwner & { + readonly clientIdentity?: string; + readonly clientManifestGeneration?: number; +}; + /** A model `resume` forwarded from another session of the same owner, carrying * the workspace-write access its own request was authenticated with. */ export type McpModelResumeCaller = McpApprovalOwner & { @@ -155,6 +164,10 @@ interface SessionMetaBase { * Pins browser-handoff URLs to the right org's console. */ readonly organizationSlug?: string; readonly userId: string; + /** Verified persistent-client identity, pinned for cold restoration. */ + readonly clientIdentity?: string; + /** Capability manifest revision, pinned for cold restoration. */ + readonly clientManifestGeneration?: number; readonly elicitationMode?: McpElicitationMode; /** Whether the session serves artifacts (carried from {@link McpSessionInit}). * Absent — including for sessions persisted before the flag existed — means @@ -1599,7 +1612,7 @@ export abstract class McpAgentSessionDOBase< } async validateMcpSessionOwner( - identity: McpApprovalOwner, + identity: McpSessionOwnerIdentity, resource: McpResource, ): Promise<"ok" | "not_found" | "forbidden" | "terminated"> { const self = this; @@ -1631,7 +1644,10 @@ export abstract class McpAgentSessionDOBase< } const ownerMatches = identity.accountId === sessionMeta.userId && - identity.organizationId === sessionMeta.organizationId; + identity.organizationId === sessionMeta.organizationId && + (identity.clientIdentity ?? null) === (sessionMeta.clientIdentity ?? null) && + (identity.clientManifestGeneration ?? null) === + (sessionMeta.clientManifestGeneration ?? null); const resourceMatches = mcpResourceKey(resource) === mcpResourceKey(sessionMeta.resource); return ownerMatches && resourceMatches ? ("ok" as const) : ("forbidden" as const); }).pipe( diff --git a/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts b/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts index 54583f2e55..030fc23efd 100644 --- a/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts +++ b/packages/hosts/cloudflare/src/mcp/execution-owner-directory.ts @@ -39,6 +39,14 @@ const toMcpExecutionOwnerDirectoryStub = (stub: unknown): McpExecutionOwnerDirec export const mcpSessionDurableObjectName = (sessionId: string): string => `streamable-http:${sessionId}`; +export const mcpClientChannelDurableObjectName = (clientId: string): string => + `mcp-client-channel:${clientId}`; + +export const mcpClientChannelIdFromName = ( + namespace: McpExecutionOwnerDirectoryNamespace, + clientId: string, +): Id => namespace.idFromName(mcpClientChannelDurableObjectName(clientId)); + class McpExecutionOwnerDirectoryRpcError extends Data.TaggedError( "McpExecutionOwnerDirectoryRpcError", )<{ diff --git a/packages/hosts/cloudflare/src/mcp/session-stub.ts b/packages/hosts/cloudflare/src/mcp/session-stub.ts index 184161e908..57478c5469 100644 --- a/packages/hosts/cloudflare/src/mcp/session-stub.ts +++ b/packages/hosts/cloudflare/src/mcp/session-stub.ts @@ -9,6 +9,7 @@ import type { McpSessionApprovalResult, McpSessionModelResumeResult, McpSessionResumeApprovalResult, + McpSessionOwnerIdentity, } from "./agent-session-durable-object"; import { mcpSessionDurableObjectName } from "./execution-owner-directory"; @@ -19,7 +20,7 @@ export interface McpSessionNamespace { export interface McpSessionStub { readonly validateMcpSessionOwner: ( - identity: McpApprovalOwner, + identity: McpSessionOwnerIdentity, resource: McpResource, ) => Promise<"ok" | "not_found" | "forbidden" | "terminated">; readonly _cf_scheduleDestroy: () => Promise; diff --git a/packages/hosts/mcp/package.json b/packages/hosts/mcp/package.json index 1e8dd7c4fc..2a6847ffae 100644 --- a/packages/hosts/mcp/package.json +++ b/packages/hosts/mcp/package.json @@ -28,6 +28,10 @@ "types": "./src/in-memory-session-store.ts", "default": "./src/in-memory-session-store.ts" }, + "./in-memory-client-bridge": { + "types": "./src/in-memory-client-bridge.ts", + "default": "./src/in-memory-client-bridge.ts" + }, "./browser-approval": { "types": "./src/browser-approval.ts", "default": "./src/browser-approval.ts" diff --git a/packages/hosts/mcp/src/envelope.test.ts b/packages/hosts/mcp/src/envelope.test.ts index 072f98fe11..0f29dacb58 100644 --- a/packages/hosts/mcp/src/envelope.test.ts +++ b/packages/hosts/mcp/src/envelope.test.ts @@ -136,6 +136,9 @@ describe("McpServingRoutes envelope", () => { expect(response.headers.get("access-control-allow-origin")).toBe("*"); expect(response.headers.get("access-control-allow-methods")).toBe("GET, POST, DELETE, OPTIONS"); expect(response.headers.get("access-control-allow-headers") ?? "").toContain("authorization"); + expect(response.headers.get("access-control-allow-headers") ?? "").toContain( + "x-executor-client-token", + ); }); it("renders 500 -32603 + CORS and fires the reporter on an orchestration defect", async () => { @@ -225,6 +228,44 @@ it("dispatches toolkit MCP routes with the parsed toolkit resource", async () => }); }); +it("forwards only the authenticated client identity into session dispatch", async () => { + const seen = await Effect.runPromise( + Ref.make<{ readonly clientIdentity?: string; readonly clientManifestGeneration?: number }>({}), + ); + const ClientAuthProviderLive = Layer.succeed(McpAuthProvider)({ + discoveryRoutes: [], + resourceMetadataUrl: (request) => `${new URL(request.url).origin}${DISCOVERY_PATH}`, + authenticate: () => Effect.succeed(authenticated(TEST_PRINCIPAL, "linked-client-a", 2)), + }); + const RecordingStoreLive = Layer.succeed(McpSessionStore)({ + dispatch: ({ clientIdentity, clientManifestGeneration }): Effect.Effect => + Ref.set(seen, { clientIdentity, clientManifestGeneration }).pipe( + Effect.as(new Response(JSON.stringify({ jsonrpc: "2.0", id: 1 }), { status: 200 })), + ), + dispose: () => Effect.void, + }); + + const handler = buildHandler(RecordingStoreLive, McpErrorReporterNoop, ClientAuthProviderLive); + const response = await handler( + new Request("https://host.test/mcp", { + method: "POST", + headers: MCP_POST_HEADERS, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 1, + method: "initialize", + params: { clientInfo: { name: "forged-client", version: "1.0.0" } }, + }), + }), + ); + + expect(response.status).toBe(200); + expect(await Effect.runPromise(Ref.get(seen))).toEqual({ + clientIdentity: "linked-client-a", + clientManifestGeneration: 2, + }); +}); + // --------------------------------------------------------------------------- // The pre-initialize dispatch guard. Session-less, only `initialize` is servable, // and the transport's answer for everything else is a connection-killing HTTP diff --git a/packages/hosts/mcp/src/envelope.ts b/packages/hosts/mcp/src/envelope.ts index 8e0b5f2252..10126789b2 100644 --- a/packages/hosts/mcp/src/envelope.ts +++ b/packages/hosts/mcp/src/envelope.ts @@ -76,7 +76,7 @@ const corsPreflightResponse = (): Response => "access-control-allow-origin": "*", "access-control-allow-methods": "GET, POST, DELETE, OPTIONS", "access-control-allow-headers": - "content-type, authorization, mcp-session-id, accept, mcp-protocol-version", + "content-type, authorization, mcp-session-id, accept, mcp-protocol-version, x-executor-client-token", "access-control-expose-headers": "mcp-session-id, WWW-Authenticate", }, }); @@ -385,6 +385,10 @@ const mcpDispatch = (resource: McpResource) => const result: McpDispatchResult = yield* store.dispatch({ request, principal, + ...(outcome.clientIdentity === undefined ? {} : { clientIdentity: outcome.clientIdentity }), + ...(outcome.clientManifestGeneration === undefined + ? {} + : { clientManifestGeneration: outcome.clientManifestGeneration }), resource, sessionId, method: request.method, diff --git a/packages/hosts/mcp/src/in-memory-client-bridge.test.ts b/packages/hosts/mcp/src/in-memory-client-bridge.test.ts new file mode 100644 index 0000000000..44c79b65a4 --- /dev/null +++ b/packages/hosts/mcp/src/in-memory-client-bridge.test.ts @@ -0,0 +1,207 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { makeInMemoryMcpClientBridge, type McpClientManifest } from "./in-memory-client-bridge"; +import type { Principal } from "./seams"; + +const principal = (accountId: string): Principal => ({ + accountId, + organizationId: "org", + organizationName: "Org", + email: `${accountId}@example.com`, + name: null, + avatarUrl: null, + roles: [], + orgRoleModel: "none", +}); + +const manifest = (name: string): McpClientManifest => ({ + tools: [{ name }], + resources: [{ uri: `file:///${name}` }], + prompts: [{ name }], +}); + +describe("in-memory MCP client bridge", () => { + it("isolates identities and rejects principal mismatches", () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("a"); + const a = bridge.connect({ principal: owner, send: () => {}, manifest: manifest("a") }); + const b = bridge.connect({ + principal: principal("b"), + send: () => {}, + manifest: manifest("b"), + }); + expect(bridge.authenticate(a.bindingToken, owner)).toEqual(expect.any(String)); + expect(bridge.authenticate(a.bindingToken, principal("b"))).toBeNull(); + expect(bridge.authenticate(b.bindingToken, owner)).toBeNull(); + expect(bridge.getManifest(a.bindingToken, owner)?.manifest.tools[0]?.name).toBe("a"); + }); + + it("routes duplicate tool names only to their originating client", async () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("shared-owner"); + const makeClient = (result: string) => { + let connection!: ReturnType; + connection = bridge.connect({ + principal: owner, + manifest: { + tools: [{ name: "filesystem.read" }, { name: `${result}.only` }], + resources: [], + prompts: [], + }, + send: (frame) => connection.respond(frame.id, result), + }); + return connection; + }; + const clientA = makeClient("client-a"); + const clientB = makeClient("client-b"); + const call = (bindingToken: string, name: string) => + bridge.request(bindingToken, owner, { + method: "tools/call", + params: { name }, + }); + + await expect(call(clientA.bindingToken, "filesystem.read")).resolves.toBe("client-a"); + await expect(call(clientB.bindingToken, "filesystem.read")).resolves.toBe("client-b"); + await expect( + bridge.requestForIdentity( + bridge.authenticate(clientA.bindingToken, owner)!, + owner, + clientA.manifestGeneration, + { method: "tools/call", params: { name: "filesystem.read" } }, + ), + ).resolves.toBe("client-a"); + await expect(call(clientA.bindingToken, "client-b.only")).rejects.toMatchObject({ + reason: "capability-not-found", + }); + await expect(call(clientB.bindingToken, "client-a.only")).rejects.toMatchObject({ + reason: "capability-not-found", + }); + }); + + it("replaces manifests without crossing client boundaries", () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("a"); + const a = bridge.connect({ principal: owner, send: () => {}, manifest: manifest("old") }); + const b = bridge.connect({ principal: owner, send: () => {}, manifest: manifest("b") }); + const nextGeneration = bridge.replaceManifest(a.bindingToken, owner, manifest("new")); + expect(nextGeneration).not.toBeNull(); + expect(nextGeneration!.manifestGeneration).toBeGreaterThan(a.manifestGeneration); + expect(bridge.getManifest(a.bindingToken, owner)?.manifest.tools[0]?.name).toBe("old"); + expect(bridge.getManifest(nextGeneration!.bindingToken, owner)?.manifest.tools[0]?.name).toBe( + "new", + ); + expect(bridge.getManifest(b.bindingToken, owner)?.manifest.tools[0]?.name).toBe("b"); + }); + + it("rejects stale pinned generations while preserving existing sessions", async () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("a"); + let oldConnection!: ReturnType; + oldConnection = bridge.connect({ + principal: owner, + manifest: manifest("old"), + send: (frame) => oldConnection.respond(frame.id, "old-result"), + }); + const replacement = bridge.replaceManifest(oldConnection.bindingToken, owner, manifest("new")); + expect(replacement).not.toBeNull(); + const nextConnection = replacement!; + + await expect( + bridge.request( + nextConnection.bindingToken, + owner, + { + method: "tools/call", + params: { name: "new" }, + }, + oldConnection.manifestGeneration, + ), + ).rejects.toMatchObject({ reason: "stale-generation" }); + expect( + await bridge.request( + oldConnection.bindingToken, + owner, + { method: "tools/call", params: { name: "old" } }, + oldConnection.manifestGeneration, + ), + ).toBe("old-result"); + }); + + it("correlates outbound tool, resource and prompt responses", async () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("a"); + const frames: Array<{ id: string; method: string }> = []; + let connection!: ReturnType; + connection = bridge.connect({ + principal: owner, + manifest: manifest("capability"), + send: (frame) => { + frames.push(frame); + connection.respond(frame.id, { method: frame.method }); + }, + }); + const methods = [ + { method: "tools/call" as const, params: { name: "capability" } }, + { method: "resources/read" as const, params: { uri: "file:///capability" } }, + { method: "prompts/get" as const, params: { name: "capability" } }, + ]; + for (const request of methods) { + expect(await bridge.request(connection.bindingToken, owner, request)).toEqual({ + method: request.method, + }); + } + expect(frames.map((frame) => frame.method)).toEqual(methods.map((request) => request.method)); + expect(new Set(frames.map((frame) => frame.id)).size).toBe(3); + }); + + it("cleans pending requests on disconnect and timeout", async () => { + const bridge = makeInMemoryMcpClientBridge({ requestTimeoutMs: 10 }); + const owner = principal("a"); + let connection!: ReturnType; + connection = bridge.connect({ + principal: owner, + manifest: manifest("a"), + send: () => {}, + }); + const disconnected = bridge.request(connection.bindingToken, owner, { + method: "tools/call", + params: { name: "a" }, + }); + connection.disconnect(); + await expect(disconnected).rejects.toThrow("disconnected"); + expect(bridge.size).toBe(0); + + const timeoutClient = bridge.connect({ + principal: owner, + send: () => {}, + manifest: manifest("a"), + }); + await expect( + bridge.request(timeoutClient.bindingToken, owner, { + method: "resources/read", + params: { uri: "file:///a" }, + }), + ).rejects.toThrow("timed out"); + }); + + it("binds tokens to stable ownership rather than mutable claims", () => { + const bridge = makeInMemoryMcpClientBridge(); + const owner = principal("a"); + const connection = bridge.connect({ + principal: owner, + send: () => {}, + manifest: manifest("a"), + }); + const refreshedPrincipal: Principal = { + ...owner, + organizationName: "Renamed Org", + email: "updated@example.com", + roles: ["admin"], + orgRoleModel: "organization", + orgRole: "admin", + }; + expect(bridge.authenticate(connection.bindingToken, refreshedPrincipal)).toEqual( + expect.any(String), + ); + }); +}); diff --git a/packages/hosts/mcp/src/in-memory-client-bridge.ts b/packages/hosts/mcp/src/in-memory-client-bridge.ts new file mode 100644 index 0000000000..1936235a3f --- /dev/null +++ b/packages/hosts/mcp/src/in-memory-client-bridge.ts @@ -0,0 +1,377 @@ +import { Data, Effect, Exit, Match } from "effect"; + +import type { Principal } from "./seams"; + +export interface McpToolManifest { + readonly name: string; + readonly description?: string; + readonly inputSchema?: Readonly>; +} + +export interface McpResourceManifest { + readonly uri: string; + readonly name?: string; + readonly description?: string; + readonly mimeType?: string; +} + +export interface McpPromptManifest { + readonly name: string; + readonly description?: string; + readonly arguments?: readonly { + readonly name: string; + readonly description?: string; + readonly required?: boolean; + }[]; +} + +/** Server-owned capabilities for one connected client. Identity is never part of this value. */ +export interface McpClientManifest { + readonly tools: readonly McpToolManifest[]; + readonly resources: readonly McpResourceManifest[]; + readonly prompts: readonly McpPromptManifest[]; +} + +export type McpClientRequest = + | { + readonly method: "tools/call"; + readonly params: { + readonly name: string; + readonly arguments?: Readonly>; + }; + } + | { readonly method: "resources/read"; readonly params: { readonly uri: string } } + | { + readonly method: "prompts/get"; + readonly params: { + readonly name: string; + readonly arguments?: Readonly>; + }; + }; + +export interface McpClientFrame { + readonly id: string; + readonly method: McpClientRequest["method"]; + readonly params: McpClientRequest["params"]; +} + +export interface McpBridgeConnection { + readonly bindingToken: string; + /** Pin this generation when creating an MCP session. */ + readonly manifestGeneration: number; + readonly respond: (id: string, result: unknown, error?: unknown) => boolean; + readonly disconnect: () => void; +} + +export interface McpClientManifestSnapshot { + readonly generation: number; + readonly manifest: McpClientManifest; +} + +/** The portion of a client bridge required to build a manifest-bound MCP session. */ +export interface McpClientSessionResolver { + readonly getManifestForIdentity: ( + clientIdentity: string, + principal: Principal, + generation: number, + ) => McpClientManifest | null; + readonly requestForIdentity: ( + clientIdentity: string, + principal: Principal, + generation: number, + request: McpClientRequest, + ) => Promise; +} + +export class McpClientBridgeError extends Data.TaggedError("McpClientBridgeError")<{ + readonly reason: + | "disconnected" + | "timeout" + | "unauthenticated" + | "capability-not-found" + | "delivery" + | "remote" + | "stale-generation"; + readonly cause?: unknown; + readonly expectedGeneration?: number; + readonly actualGeneration?: number; +}> { + override get message(): string { + return Match.value(this.reason).pipe( + Match.when("disconnected", () => "MCP client disconnected"), + Match.when("timeout", () => "MCP client request timed out"), + Match.when("unauthenticated", () => "MCP client not authenticated"), + Match.when("capability-not-found", () => "MCP capability not found"), + Match.when("delivery", () => "MCP frame delivery failed"), + Match.when("remote", () => "MCP client returned an error"), + Match.when( + "stale-generation", + () => + `MCP session generation is stale (expected ${this.expectedGeneration}, current ${this.actualGeneration})`, + ), + Match.exhaustive, + ); + } +} + +interface PendingRequest { + readonly resume: (effect: Effect.Effect) => void; + readonly timer: ReturnType; +} + +interface ClientRecord { + readonly principal: Principal; + readonly clientIdentity: string; + readonly send: (frame: McpClientFrame) => void; + readonly manifest: McpClientManifest; + readonly manifestGeneration: number; + active: boolean; + readonly pending: Map; +} + +const owns = (owner: Principal, principal: Principal): boolean => + owner.accountId === principal.accountId && owner.organizationId === principal.organizationId; + +const opaqueId = (): string => globalThis.crypto.randomUUID(); + +/** + * In-memory, transport-independent bridge to authenticated MCP clients. + * The transport adapter owns delivering frames and calling `respond`. + */ +export const makeInMemoryMcpClientBridge = ( + options: { readonly requestTimeoutMs?: number } = {}, +) => { + const timeoutMs = options.requestTimeoutMs ?? 30_000; + const clients = new Map(); + let latestManifestGeneration = 0; + + const connectionFor = (token: string, record: ClientRecord): McpBridgeConnection => ({ + bindingToken: token, + manifestGeneration: record.manifestGeneration, + respond: (id, result, error) => { + const pending = record.pending.get(id); + if (!record.active || !pending) return false; + record.pending.delete(id); + clearTimeout(pending.timer); + if (error !== undefined) { + pending.resume(Effect.fail(new McpClientBridgeError({ reason: "remote", cause: error }))); + } else { + pending.resume(Effect.succeed(result)); + } + return true; + }, + disconnect: () => disconnect(token, record), + }); + + const disconnect = (token: string, record: ClientRecord): void => { + if (!record.active) return; + record.active = false; + clients.delete(token); + for (const pending of record.pending.values()) { + clearTimeout(pending.timer); + pending.resume(Effect.fail(new McpClientBridgeError({ reason: "disconnected" }))); + } + record.pending.clear(); + }; + + const route = (record: ClientRecord, request: McpClientRequest): Promise => { + if (!record.active) { + return Effect.runPromise(Effect.fail(new McpClientBridgeError({ reason: "disconnected" }))); + } + const id = opaqueId(); + return Effect.runPromise( + Effect.callback((resume) => { + const timer = setTimeout(() => { + record.pending.delete(id); + resume(Effect.fail(new McpClientBridgeError({ reason: "timeout" }))); + }, timeoutMs); + record.pending.set(id, { resume, timer }); + const sent = Effect.runSyncExit( + Effect.try({ + try: () => record.send({ id, method: request.method, params: request.params }), + catch: (cause) => new McpClientBridgeError({ reason: "delivery", cause }), + }), + ); + if (Exit.isFailure(sent)) { + clearTimeout(timer); + record.pending.delete(id); + resume(Effect.fail(new McpClientBridgeError({ reason: "delivery" }))); + } + return Effect.sync(() => { + clearTimeout(timer); + record.pending.delete(id); + }); + }), + ); + }; + + return { + connect(input: { + readonly principal: Principal; + readonly send: (frame: McpClientFrame) => void; + readonly manifest: McpClientManifest; + }): McpBridgeConnection { + const token = opaqueId() + opaqueId(); + const record: ClientRecord = { + principal: input.principal, + clientIdentity: opaqueId(), + send: input.send, + manifest: input.manifest, + manifestGeneration: ++latestManifestGeneration, + active: true, + pending: new Map(), + }; + clients.set(token, record); + return connectionFor(token, record); + }, + authenticate(token: string, principal: Principal): string | null { + const record = clients.get(token); + return record?.active && owns(record.principal, principal) ? record.clientIdentity : null; + }, + /** Resolve the server-owned identity and pinned manifest generation for a binding. */ + getBinding( + token: string, + principal: Principal, + ): { readonly clientIdentity: string; readonly manifestGeneration: number } | null { + const record = clients.get(token); + return record?.active && owns(record.principal, principal) + ? { + clientIdentity: record.clientIdentity, + manifestGeneration: record.manifestGeneration, + } + : null; + }, + /** + * Create a new session generation with the replacement manifest. The + * existing token and its manifest remain pinned to their original + * generation; callers must initialize a new MCP session using the returned + * connection to observe the update. Supply a generation-specific `send` + * callback when the new transport session uses a different outbound route. + */ + replaceManifest( + token: string, + principal: Principal, + manifest: McpClientManifest, + send?: (frame: McpClientFrame) => void, + ): McpBridgeConnection | null { + const record = clients.get(token); + if (!record?.active || !owns(record.principal, principal)) return null; + const nextToken = opaqueId() + opaqueId(); + const nextRecord: ClientRecord = { + principal: record.principal, + clientIdentity: record.clientIdentity, + send: send ?? record.send, + manifest, + manifestGeneration: ++latestManifestGeneration, + active: true, + pending: new Map(), + }; + clients.set(nextToken, nextRecord); + return connectionFor(nextToken, nextRecord); + }, + getManifest(token: string, principal: Principal): McpClientManifestSnapshot | null { + const record = clients.get(token); + return record?.active && owns(record.principal, principal) + ? { generation: record.manifestGeneration, manifest: record.manifest } + : null; + }, + getManifestForIdentity( + clientIdentity: string, + principal: Principal, + generation: number, + ): McpClientManifest | null { + for (const record of clients.values()) { + if ( + record.active && + record.clientIdentity === clientIdentity && + record.manifestGeneration === generation && + owns(record.principal, principal) + ) { + return record.manifest; + } + } + return null; + }, + /** + * Pass the generation pinned when the MCP session was initialized. A + * mismatch fails closed as stale; rebuild the MCP session from its + * replacement connection rather than changing an existing session in place. + */ + request( + token: string, + principal: Principal, + request: McpClientRequest, + expectedManifestGeneration?: number, + ): Promise { + const record = clients.get(token); + if (!record?.active || !owns(record.principal, principal)) { + return Effect.runPromise( + Effect.fail(new McpClientBridgeError({ reason: "unauthenticated" })), + ); + } + if ( + expectedManifestGeneration !== undefined && + expectedManifestGeneration !== record.manifestGeneration + ) { + return Effect.runPromise( + Effect.fail( + new McpClientBridgeError({ + reason: "stale-generation", + expectedGeneration: expectedManifestGeneration, + actualGeneration: record.manifestGeneration, + }), + ), + ); + } + const known = + request.method === "tools/call" + ? record.manifest.tools.some((item) => item.name === request.params.name) + : request.method === "resources/read" + ? record.manifest.resources.some((item) => item.uri === request.params.uri) + : record.manifest.prompts.some((item) => item.name === request.params.name); + if (!known) + return Effect.runPromise( + Effect.fail(new McpClientBridgeError({ reason: "capability-not-found" })), + ); + return route(record, request); + }, + requestForIdentity( + clientIdentity: string, + principal: Principal, + generation: number, + request: McpClientRequest, + ): Promise { + const record = [...clients.values()].find( + (candidate) => + candidate.active && + candidate.clientIdentity === clientIdentity && + candidate.manifestGeneration === generation && + owns(candidate.principal, principal), + ); + if (!record) { + return Effect.runPromise( + Effect.fail(new McpClientBridgeError({ reason: "unauthenticated" })), + ); + } + const known = + request.method === "tools/call" + ? record.manifest.tools.some((item) => item.name === request.params.name) + : request.method === "resources/read" + ? record.manifest.resources.some((item) => item.uri === request.params.uri) + : record.manifest.prompts.some((item) => item.name === request.params.name); + if (!known) { + return Effect.runPromise( + Effect.fail(new McpClientBridgeError({ reason: "capability-not-found" })), + ); + } + return route(record, request); + }, + disconnect(token: string): void { + const record = clients.get(token); + if (record) disconnect(token, record); + }, + get size(): number { + return clients.size; + }, + }; +}; diff --git a/packages/hosts/mcp/src/in-memory-session-store.test.ts b/packages/hosts/mcp/src/in-memory-session-store.test.ts index bdc51db85d..8c51c6cd07 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.test.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.test.ts @@ -11,6 +11,8 @@ import { type McpBuildServer, type McpBuildServerOptions, } from "./in-memory-session-store"; +import { makeInMemoryMcpClientBridge } from "./in-memory-client-bridge"; +import type { McpClientSessionResolver } from "./in-memory-client-bridge"; import { defaultMcpResource, type Principal } from "./seams"; import { createExecutorMcpServer } from "./tool-server"; @@ -25,6 +27,11 @@ const TEST_PRINCIPAL: Principal = { orgRoleModel: "organization", }; +const TEST_CLIENT_RESOLVER: McpClientSessionResolver = { + getManifestForIdentity: () => ({ tools: [], resources: [], prompts: [] }), + requestForIdentity: () => Promise.resolve(null), +}; + it("preserves native elicitation mode when creating an in-memory MCP session", async () => { let buildOptions: McpBuildServerOptions | undefined; const sessions = makeInMemoryMcpSessionStore((_principal, options) => { @@ -60,6 +67,237 @@ it("preserves native elicitation mode when creating an in-memory MCP session", a expect(buildOptions?.elicitationMode).toEqual({ mode: "native" }); }); +it("fails closed when an authenticated client binding has no exact manifest", async () => { + let buildCalled = false; + const sessions = makeInMemoryMcpSessionStore(() => { + buildCalled = true; + return Effect.fail(new McpEngineBuildError({ cause: "should not build" })); + }); + + const response = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "initialize", params: {} }), + }), + principal: TEST_PRINCIPAL, + clientIdentity: "client-a", + clientManifestGeneration: 1, + resource: defaultMcpResource, + sessionId: null, + method: "POST", + }), + ); + + expect(response).toBeInstanceOf(Response); + expect((response as Response).status).toBe(403); + expect(buildCalled).toBe(false); +}); + +it("rejects reuse of an MCP session by a different authenticated client", async () => { + const sessions = makeInMemoryMcpSessionStore( + () => + createExecutorMcpServer({ engine: makeIdleTestEngine() }).pipe( + Effect.map((mcpServer) => ({ mcpServer, engine: makeIdleTestEngine() })), + ), + { clientSessionResolver: TEST_CLIENT_RESOLVER }, + ); + const sessionId = await openSession( + sessions, + TEST_PRINCIPAL, + "https://executor.test/mcp", + "client-a", + ); + + const result = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + ...MCP_POST_HEADERS, + "mcp-session-id": sessionId, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 2, + method: "tools/list", + }), + }), + principal: TEST_PRINCIPAL, + clientIdentity: "client-b", + resource: defaultMcpResource, + sessionId, + method: "POST", + }), + ); + + expect(result).toBe("forbidden"); + + const unlinkedResult = await Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + ...MCP_POST_HEADERS, + "mcp-session-id": sessionId, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 3, + method: "tools/list", + }), + }), + principal: TEST_PRINCIPAL, + resource: defaultMcpResource, + sessionId, + method: "POST", + }), + ); + + expect(unlinkedResult).toBe("forbidden"); + await sessions.close(); +}); + +it("builds each session with only its authenticated client identity", async () => { + const identities: Array = []; + const sessions = makeInMemoryMcpSessionStore( + (_principal, options) => { + identities.push(options?.clientIdentity); + return createExecutorMcpServer({ engine: makeIdleTestEngine() }).pipe( + Effect.map((mcpServer) => ({ mcpServer, engine: makeIdleTestEngine() })), + ); + }, + { clientSessionResolver: TEST_CLIENT_RESOLVER }, + ); + + await openSession(sessions, TEST_PRINCIPAL, "https://executor.test/mcp", "client-a"); + await openSession(sessions, TEST_PRINCIPAL, "https://executor.test/mcp", "client-b"); + + expect(identities).toEqual(["client-a", "client-b"]); + await sessions.close(); +}); + +it("requires a new MCP session after a linked client's manifest changes", async () => { + const sessions = makeInMemoryMcpSessionStore( + () => + createExecutorMcpServer({ engine: makeIdleTestEngine() }).pipe( + Effect.map((mcpServer) => ({ mcpServer, engine: makeIdleTestEngine() })), + ), + { clientSessionResolver: TEST_CLIENT_RESOLVER }, + ); + const sessionId = await openSession( + sessions, + TEST_PRINCIPAL, + "https://executor.test/mcp", + "client-a", + 1, + ); + + const requestForGeneration = (clientManifestGeneration: number) => + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + ...MCP_POST_HEADERS, + "mcp-session-id": sessionId, + }, + body: JSON.stringify({ jsonrpc: "2.0", id: 2, method: "tools/list" }), + }), + principal: TEST_PRINCIPAL, + clientIdentity: "client-a", + clientManifestGeneration, + resource: defaultMcpResource, + sessionId, + method: "POST", + }); + + expect(await Effect.runPromise(requestForGeneration(1))).toBeInstanceOf(Response); + expect(await Effect.runPromise(requestForGeneration(2))).toBe("forbidden"); + await sessions.close(); +}); + +it("routes same-named manifest tools to their originating client only", async () => { + const bridge = makeInMemoryMcpClientBridge(); + const makeClient = (label: string) => { + let connection: ReturnType; + connection = bridge.connect({ + principal: TEST_PRINCIPAL, + manifest: { + tools: [{ name: "client_tool", description: `${label} tool` }], + resources: [], + prompts: [], + }, + send: (frame) => { + connection.respond(frame.id, { content: [{ type: "text", text: label }] }); + }, + }); + return { + identity: bridge.getBinding(connection.bindingToken, TEST_PRINCIPAL)!.clientIdentity, + generation: connection.manifestGeneration, + }; + }; + const clientA = makeClient("A"); + const clientB = makeClient("B"); + const sessions = makeInMemoryMcpSessionStore( + (_principal, options) => + createExecutorMcpServer({ + engine: makeIdleTestEngine(), + ...(options?.clientManifest === undefined + ? {} + : { clientManifest: options.clientManifest }), + ...(options?.invokeClient === undefined ? {} : { invokeClient: options.invokeClient }), + }).pipe(Effect.map((mcpServer) => ({ mcpServer, engine: makeIdleTestEngine() }))), + { clientSessionResolver: bridge }, + ); + const sessionA = await openSession( + sessions, + TEST_PRINCIPAL, + "https://executor.test/mcp", + clientA.identity, + clientA.generation, + ); + const sessionB = await openSession( + sessions, + TEST_PRINCIPAL, + "https://executor.test/mcp", + clientB.identity, + clientB.generation, + ); + const call = async (sessionId: string, identity: string, generation: number) => + Effect.runPromise( + sessions.store.dispatch({ + request: new Request("https://executor.test/mcp", { + method: "POST", + headers: { + ...MCP_POST_HEADERS, + "mcp-session-id": sessionId, + }, + body: JSON.stringify({ + jsonrpc: "2.0", + id: 7, + method: "tools/call", + params: { name: "client_tool", arguments: {} }, + }), + }), + principal: TEST_PRINCIPAL, + clientIdentity: identity, + clientManifestGeneration: generation, + resource: defaultMcpResource, + sessionId, + method: "POST", + }), + ); + + const [responseA, responseB] = await Promise.all([ + call(sessionA, clientA.identity, clientA.generation), + call(sessionB, clientB.identity, clientB.generation), + ]); + expect(await (responseA as Response).text()).toContain('"text":"A"'); + expect(await (responseB as Response).text()).toContain('"text":"B"'); + await sessions.close(); +}); + /** A do-nothing engine: the eviction test drives session lifetime, not tools. */ const makeIdleTestEngine = (): ExecutionEngine => ({ execute: () => Effect.succeed({ result: "unused" }), @@ -127,6 +365,8 @@ const openSession = async ( sessions: TestSessionStore, principal: Principal = TEST_PRINCIPAL, requestUrl = "https://executor.test/mcp", + clientIdentity?: string, + clientManifestGeneration?: number, ): Promise => { const response = (await Effect.runPromise( sessions.store.dispatch({ @@ -148,6 +388,10 @@ const openSession = async ( }), }), principal, + ...(clientIdentity === undefined ? {} : { clientIdentity }), + ...(clientIdentity === undefined + ? {} + : { clientManifestGeneration: clientManifestGeneration ?? 1 }), resource: defaultMcpResource, sessionId: null, method: "POST", diff --git a/packages/hosts/mcp/src/in-memory-session-store.ts b/packages/hosts/mcp/src/in-memory-session-store.ts index 952f9bd83f..5ebb24a23d 100644 --- a/packages/hosts/mcp/src/in-memory-session-store.ts +++ b/packages/hosts/mcp/src/in-memory-session-store.ts @@ -20,6 +20,11 @@ import { type InProcessBrowserApprovalStore, } from "./browser-approval-store"; import { jsonRpcErrorBody, preInitializeMethodNotFound } from "./envelope"; +import type { + McpClientManifest, + McpClientRequest, + McpClientSessionResolver, +} from "./in-memory-client-bridge"; import { McpSessionStore, MCP_ORG_WRITE_ACCESS_HEADER, @@ -107,6 +112,14 @@ export interface BuiltMcpServer { /** The browser-mode wiring the store hands a build call when a session opts in. */ export interface McpBuildServerOptions { readonly resource?: McpResource; + /** Authenticated client identity used to select session-scoped capabilities. */ + readonly clientIdentity?: string; + /** Manifest revision captured when this MCP session was initialized. */ + readonly clientManifestGeneration?: number; + /** Exact manifest snapshot selected by the authenticated binding. */ + readonly clientManifest?: McpClientManifest; + /** Routes capability calls back to the authenticated originating client. */ + readonly invokeClient?: (request: McpClientRequest) => Promise; readonly elicitationMode?: | { readonly mode: "browser"; readonly approvalUrl: (executionId: string) => string } | { readonly mode: "model" } @@ -213,15 +226,21 @@ const RESUME_PATH = /^\/api\/mcp-sessions\/([^/?#]+)\/executions\/([^/?#]+)\/res interface SessionOwner { readonly principal: Principal; + readonly clientIdentity: string | null; + readonly clientManifestGeneration: number | null; readonly resource: McpResource; } const sessionOwnerMatches = ( owner: SessionOwner, principal: Principal, + clientIdentity: string | undefined, + clientManifestGeneration: number | undefined, resource: McpResource, ): boolean => principalOwns(owner.principal, principal) && + owner.clientIdentity === (clientIdentity ?? null) && + owner.clientManifestGeneration === (clientManifestGeneration ?? null) && mcpResourceKey(owner.resource) === mcpResourceKey(resource); /** @@ -243,6 +262,8 @@ export const makeInMemoryMcpSessionStore = ( readonly sessionIdleTtlMs?: number; /** How often the sweep runs. Defaults to a quarter of the TTL. */ readonly sessionSweepIntervalMs?: number; + /** Resolver for authenticated, manifest-bound client sessions. */ + readonly clientSessionResolver?: McpClientSessionResolver; } = {}, ): InMemoryMcpSessionStore => { const transports = new Map(); @@ -387,14 +408,22 @@ export const makeInMemoryMcpSessionStore = ( const forward = ( sessionId: string, principal: Principal, + clientIdentity: string | undefined, + clientManifestGeneration: number | undefined, resource: McpResource, request: Request, ): Effect.Effect => { const transport = transports.get(sessionId); const owner = owners.get(sessionId); if (!transport || !owner) return Effect.succeed("not-found"); - if (!sessionOwnerMatches(owner, principal, resource)) return Effect.succeed("forbidden"); - owners.set(sessionId, { principal, resource }); + if (!sessionOwnerMatches(owner, principal, clientIdentity, clientManifestGeneration, resource)) + return Effect.succeed("forbidden"); + owners.set(sessionId, { + principal, + clientIdentity: clientIdentity ?? null, + clientManifestGeneration: clientManifestGeneration ?? null, + resource, + }); touch(sessionId); // Claim before the await, release in the finalizer — `runHandleRequest` // already recovers every failure to a 500, but `ensuring` also covers an @@ -448,12 +477,50 @@ export const makeInMemoryMcpSessionStore = ( /** Open a new session: build the server, connect a transport, drive the request. */ const openSession = ( principal: Principal, + clientIdentity: string | undefined, + clientManifestGeneration: number | undefined, resource: McpResource, request: Request, ): Effect.Effect => { let createdSessionId: string | null = null; + const clientResolver = options.clientSessionResolver; + const clientManifest = + clientIdentity !== undefined && clientManifestGeneration !== undefined + ? clientResolver?.getManifestForIdentity( + clientIdentity, + principal, + clientManifestGeneration, + ) + : undefined; + const invokeClient = + clientIdentity !== undefined && + clientManifestGeneration !== undefined && + clientResolver !== undefined + ? (clientRequest: McpClientRequest) => + clientResolver.requestForIdentity( + clientIdentity, + principal, + clientManifestGeneration, + clientRequest, + ) + : undefined; + if ( + clientIdentity !== undefined && + (clientManifestGeneration === undefined || + clientResolver === undefined || + clientManifest === null || + clientManifest === undefined || + invokeClient === undefined) + ) { + return Effect.succeed(jsonRpcError(403, -32003, "Client binding unavailable")); + } return buildServer(principal, { ...buildOptionsFor(request, () => createdSessionId), + ...(clientIdentity === undefined ? {} : { clientIdentity }), + ...(clientManifestGeneration === undefined ? {} : { clientManifestGeneration }), + ...(clientManifest === undefined || clientManifest === null + ? {} + : { clientManifest, invokeClient }), resource, }).pipe( Effect.flatMap(({ mcpServer, engine, executor, close }) => @@ -465,7 +532,12 @@ export const makeInMemoryMcpSessionStore = ( createdSessionId = sid; transports.set(sid, transport); servers.set(sid, mcpServer); - owners.set(sid, { principal, resource }); + owners.set(sid, { + principal, + clientIdentity: clientIdentity ?? null, + clientManifestGeneration: clientManifestGeneration ?? null, + resource, + }); engines.set(sid, engine); if (executor) executors.set(sid, executor); if (close) closers.set(sid, close); @@ -515,6 +587,8 @@ export const makeInMemoryMcpSessionStore = ( */ const create = ( principal: Principal, + clientIdentity: string | undefined, + clientManifestGeneration: number | undefined, resource: McpResource, request: Request, ): Effect.Effect => @@ -522,15 +596,28 @@ export const makeInMemoryMcpSessionStore = ( Effect.flatMap((unsupported) => unsupported ? Effect.succeed(unsupported) - : openSession(principal, resource, request), + : openSession(principal, clientIdentity, clientManifestGeneration, resource, request), ), ); const store: McpSessionStore["Service"] = { - dispatch: ({ request, principal, resource, sessionId }: McpDispatchInput) => + dispatch: ({ + request, + principal, + clientIdentity, + clientManifestGeneration, + resource, + sessionId, + }: McpDispatchInput) => sessionId - ? forward(sessionId, principal, resource, request) - : create(principal, resource ?? defaultMcpResource, request), + ? forward(sessionId, principal, clientIdentity, clientManifestGeneration, resource, request) + : create( + principal, + clientIdentity, + clientManifestGeneration, + resource ?? defaultMcpResource, + request, + ), dispose: (sessionId) => Effect.promise(() => dispose(sessionId, { transport: true, server: true })), }; diff --git a/packages/hosts/mcp/src/seams.ts b/packages/hosts/mcp/src/seams.ts index 01d0994db0..fb9ba6347f 100644 --- a/packages/hosts/mcp/src/seams.ts +++ b/packages/hosts/mcp/src/seams.ts @@ -142,6 +142,11 @@ export const mcpResourceKey = (resource: McpResource | null | undefined): string export interface McpAuthenticated { readonly _tag: "Authenticated"; readonly principal: Principal; + /** Verified client installation identity, if a linked client token was + * presented and validated by the host auth provider. */ + readonly clientIdentity?: string; + /** Manifest revision captured by the linked client at session initialization. */ + readonly clientManifestGeneration?: number; } export interface McpUnauthorized { @@ -171,9 +176,15 @@ export interface McpUnavailable { export type AuthOutcome = McpAuthenticated | McpUnauthorized | McpForbidden | McpUnavailable; -export const authenticated = (principal: Principal): McpAuthenticated => ({ +export const authenticated = ( + principal: Principal, + clientIdentity?: string, + clientManifestGeneration?: number, +): McpAuthenticated => ({ _tag: "Authenticated", principal, + ...(clientIdentity === undefined ? {} : { clientIdentity }), + ...(clientManifestGeneration === undefined ? {} : { clientManifestGeneration }), }); export const unauthorized = (challenge?: string): McpUnauthorized => @@ -268,6 +279,11 @@ export class McpAuthProvider extends Context.Service< export interface McpDispatchInput { readonly request: Request; readonly principal: Principal; + /** Server-verified installation identity derived from the client's linked + * persistent connection; never copied from client-controlled MCP metadata. */ + readonly clientIdentity?: string; + /** Pinned manifest revision; an update requires a fresh MCP session. */ + readonly clientManifestGeneration?: number; readonly resource: McpResource; readonly sessionId: string | null; readonly method: string; diff --git a/packages/hosts/mcp/src/tool-server.test.ts b/packages/hosts/mcp/src/tool-server.test.ts index 04fd4fca28..cfbae186ce 100644 --- a/packages/hosts/mcp/src/tool-server.test.ts +++ b/packages/hosts/mcp/src/tool-server.test.ts @@ -8,13 +8,21 @@ import type { ClientCapabilities } from "@modelcontextprotocol/sdk/types.js"; import type * as Cause from "effect/Cause"; import { + AuthTemplateSlug, + ConnectionName, ElicitationId, FormElicitation, + IntegrationSlug, + ProviderItemId, + ProviderKey, ToolAddress, + ToolName, ToolResult, UrlElicitation, + definePlugin, } from "@executor-js/sdk"; -import type { ToolFileValue } from "@executor-js/sdk"; +import type { CredentialProvider, ToolFileValue } from "@executor-js/sdk"; +import { makeTestExecutor } from "@executor-js/sdk/testing"; import type { ExecutionEngine, ExecutionResult } from "@executor-js/execution"; import { @@ -68,6 +76,8 @@ type TestServerConfig = Pick< | "pausedExecutionHooks" | "pausedExecutionLeaseMs" | "resumeFallback" + | "clientManifest" + | "invokeClient" >; /** Connect a real MCP Client to our executor MCP server over in-memory transports. */ @@ -2140,4 +2150,238 @@ describe("formatMcpExecutionOutcome serialization cost", () => { if (returned.type !== "text") return; expect(returned.text).toContain('"data"'); }); + + it("serves client tools, resources, and prompts from the session manifest", async () => { + const routed: string[] = []; + await withClient( + makeStubEngine({}), + {}, + async (client) => { + const tools = await client.listTools(); + expect(tools.tools.some((tool) => tool.name === "client_echo")).toBe(true); + const called = await client.callTool({ + name: "client_echo", + arguments: { value: "hello" }, + }); + expect(called.content).toEqual([{ type: "text", text: "hello" }]); + const resources = await client.listResources(); + expect(resources.resources.some((resource) => resource.uri === "client://item")).toBe(true); + const read = await client.readResource({ uri: "client://item" }); + expect(read.contents[0]).toMatchObject({ uri: "client://item", text: "resource" }); + const prompts = await client.listPrompts(); + expect(prompts.prompts.some((prompt) => prompt.name === "client_prompt")).toBe(true); + const prompt = await client.getPrompt({ name: "client_prompt", arguments: { topic: "x" } }); + expect(prompt.messages[0]).toMatchObject({ + role: "user", + content: { type: "text", text: "x" }, + }); + }, + { + clientManifest: { + tools: [ + { + name: "client_echo", + inputSchema: { type: "object", properties: { value: { type: "string" } } }, + }, + ], + resources: [{ uri: "client://item" }], + prompts: [{ name: "client_prompt", arguments: [{ name: "topic" }] }], + }, + invokeClient: async (request) => { + routed.push(request.method); + if (request.method === "tools/call") { + return { content: [{ type: "text", text: request.params.arguments?.value }] }; + } + if (request.method === "resources/read") { + return { contents: [{ uri: request.params.uri, text: "resource" }] }; + } + return { + messages: [ + { + role: "user", + content: { type: "text", text: request.params.arguments?.topic ?? "" }, + }, + ], + }; + }, + }, + ); + expect(routed).toEqual(["tools/call", "resources/read", "prompts/get"]); + }); + + it("keeps same-named client tools confined to their server session", async () => { + const callWith = async (text: string) => { + let routed = false; + await withClient( + makeStubEngine({}), + {}, + async (client) => { + const result = await client.callTool({ name: "shared_name", arguments: {} }); + expect(result.content).toEqual([{ type: "text", text }]); + expect( + await client + .listTools() + .then((listed) => listed.tools.some((tool) => tool.name === "only_other_session")), + ).toBe(false); + }, + { + clientManifest: { tools: [{ name: "shared_name" }], resources: [], prompts: [] }, + invokeClient: async () => { + routed = true; + return { content: [{ type: "text", text }] }; + }, + }, + ); + expect(routed).toBe(true); + }; + await callWith("session A"); + await callWith("session B"); + }); + + it("rejects calls for capabilities absent from the session manifest", async () => { + let routed = false; + await withClient( + makeStubEngine({}), + {}, + async (client) => { + const response = await client.callTool({ name: "hidden_tool", arguments: {} }); + expect(response.isError).toBe(true); + }, + { + clientManifest: { tools: [], resources: [], prompts: [] }, + invokeClient: async () => { + routed = true; + return { content: [] }; + }, + }, + ); + expect(routed).toBe(false); + }); + + it.effect("hides omitted client tools while keeping common Executor tools available", () => + Effect.scoped( + Effect.gen(function* () { + const integration = IntegrationSlug.make("shared"); + const toolCode = `return await tools["shared.org.main.read"]({ value: "hello" });`; + const executed: string[] = []; + const commonCalls: unknown[] = []; + let routedToClient = false; + const credentials = new Map(); + const commonProvider: CredentialProvider = { + key: ProviderKey.make("shared-tool-test-memory"), + writable: true, + get: (id) => Effect.sync(() => credentials.get(String(id)) ?? null), + set: (id, value) => + Effect.sync(() => { + credentials.set(String(id), value); + }), + has: (id) => Effect.sync(() => credentials.has(String(id))), + list: () => + Effect.sync(() => + Array.from(credentials.keys()).map((id) => ({ + id: ProviderItemId.make(id), + name: id, + })), + ), + }; + const commonPlugin = definePlugin(() => ({ + id: "shared-tool-test", + credentialProviders: [commonProvider], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [ + { + name: ToolName.make("read"), + description: "Read from the shared integration", + inputSchema: { + type: "object", + properties: { value: { type: "string" } }, + required: ["value"], + }, + }, + ], + }), + invokeTool: ({ toolRow, args }) => + Effect.sync(() => { + const call = { name: String(toolRow.name), args }; + commonCalls.push(call); + return { source: "shared integration", args }; + }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: integration, + description: "Shared integration", + config: {}, + }), + }), + }))(); + const executor = yield* makeTestExecutor({ + plugins: [commonPlugin] as const, + }); + yield* executor["shared-tool-test"].seed(); + yield* executor.connections.create({ + owner: "org", + name: ConnectionName.make("main"), + integration, + template: AuthTemplateSlug.make("apiKey"), + value: "test-token", + }); + const commonToolAddress = ToolAddress.make("tools.shared.org.main.read"); + const availableTools = yield* executor.tools.list(); + expect(availableTools.map((tool) => tool.address)).toContain(commonToolAddress); + const engine = makeStubEngine({ + executeWithPause: (code) => { + executed.push(code); + return executor.execute(commonToolAddress, { value: "hello" }).pipe( + Effect.map((result) => ({ + status: "completed" as const, + result: { result, logs: [] }, + })), + ); + }, + }); + + yield* Effect.promise(() => + withClient( + engine, + NO_CAPS, + async (client) => { + const tools = await client.listTools(); + expect(tools.tools.some((tool) => tool.name === "execute")).toBe(true); + expect(tools.tools.some((tool) => tool.name === "client_hidden")).toBe(false); + + const hidden = await client.callTool({ name: "client_hidden", arguments: {} }); + expect(hidden.isError).toBe(true); + + const common = await client.callTool({ + name: "execute", + arguments: { code: toolCode }, + }); + expect(executed).toEqual([toolCode]); + expect(common.isError).toBeFalsy(); + expect(common.structuredContent).toEqual( + expect.objectContaining({ + result: { source: "shared integration", args: { value: "hello" } }, + }), + ); + expect(commonCalls).toEqual([{ name: "read", args: { value: "hello" } }]); + }, + { + clientManifest: { tools: [], resources: [], prompts: [] }, + invokeClient: async () => { + routedToClient = true; + return { content: [] }; + }, + }, + ), + ).pipe(Effect.ensuring(engine.shutdown)); + + expect(routedToClient).toBe(false); + expect(executed).toEqual([toolCode]); + expect(commonCalls).toEqual([{ name: "read", args: { value: "hello" } }]); + }), + ), + ); }); diff --git a/packages/hosts/mcp/src/tool-server.ts b/packages/hosts/mcp/src/tool-server.ts index 91562be746..25147aeefa 100644 --- a/packages/hosts/mcp/src/tool-server.ts +++ b/packages/hosts/mcp/src/tool-server.ts @@ -92,6 +92,7 @@ import { SEARCH_INVOKE_SKILL, } from "./passthrough-tools"; import type { McpToolMode } from "./browser-approval"; +import type { McpClientManifest, McpClientRequest } from "./in-memory-client-bridge"; // --------------------------------------------------------------------------- // Workers-compatible JSON Schema validator (replaces Ajv which uses new Function()) @@ -111,11 +112,87 @@ class CfWorkerJsonSchemaValidator implements jsonSchemaValidator { } } +const normalizeClientResult = ( + value: unknown, +): Effect.Effect<{ content: ContentBlock[]; isError?: boolean }, McpPassthroughUnavailableError> => + Effect.gen(function* () { + if ( + typeof value !== "object" || + value === null || + !("content" in value) || + !Array.isArray(value.content) + ) { + return yield* new McpPassthroughUnavailableError({ + reason: "Client tool result must contain a content array", + }); + } + return { + content: value.content as ContentBlock[], + ...("isError" in value && typeof value.isError === "boolean" + ? { isError: value.isError } + : {}), + }; + }); + +const normalizeClientResourceResult = ( + value: unknown, + _uri: string, +): Effect.Effect< + { + contents: ( + | { uri: string; text: string; mimeType?: string } + | { uri: string; blob: string; mimeType?: string } + )[]; + }, + McpPassthroughUnavailableError +> => + Effect.gen(function* () { + if ( + typeof value !== "object" || + value === null || + !("contents" in value) || + !Array.isArray(value.contents) + ) { + return yield* new McpPassthroughUnavailableError({ + reason: "Client resource result must contain a contents array", + }); + } + return { + contents: value.contents as ( + | { uri: string; text: string; mimeType?: string } + | { uri: string; blob: string; mimeType?: string } + )[], + }; + }); + +const normalizeClientPromptResult = ( + value: unknown, +): Effect.Effect< + { messages: { role: "user" | "assistant"; content: ContentBlock }[] }, + McpPassthroughUnavailableError +> => + Effect.gen(function* () { + if ( + typeof value !== "object" || + value === null || + !("messages" in value) || + !Array.isArray(value.messages) + ) { + return yield* new McpPassthroughUnavailableError({ + reason: "Client prompt result must contain a messages array", + }); + } + return { messages: value.messages as { role: "user" | "assistant"; content: ContentBlock }[] }; + }); + // --------------------------------------------------------------------------- // Config // --------------------------------------------------------------------------- type SharedMcpServerConfig = { + /** Session-scoped capabilities routed to the authenticated client. */ + readonly clientManifest?: McpClientManifest; + readonly invokeClient?: (request: McpClientRequest) => Promise; /** * Pre-built `execute` tool description. When provided, the factory skips * its internal `engine.getDescription` yield. Useful when the caller @@ -2916,5 +2993,120 @@ export const createExecutorMcpServer = ( server.server.oninitialized = syncToolAvailability; }).pipe(Effect.withSpan("mcp.host.sync_tool_availability")); + if (config.clientManifest) { + if (!config.invokeClient) { + return yield* new McpPassthroughUnavailableError({ + reason: "client manifest requires an invocation callback", + }); + } + const manifest = config.clientManifest; + const assertUnique = (kind: string, names: readonly string[]) => + Effect.gen(function* () { + const seen = new Set(); + for (const name of names) { + if (!name.trim() || seen.has(name)) { + return yield* new McpPassthroughUnavailableError({ + reason: `Invalid client ${kind} name: ${name}`, + }); + } + seen.add(name); + } + }); + yield* assertUnique( + "tool", + manifest.tools.map((item) => item.name), + ); + yield* assertUnique( + "resource", + manifest.resources.map((item) => item.uri), + ); + yield* assertUnique( + "prompt", + manifest.prompts.map((item) => item.name), + ); + const invokeClient = config.invokeClient; + const clientContext = yield* Effect.context(); + const registeredCapabilities = [ + ["tool", manifest.tools.map((item) => item.name), Reflect.get(server, "_registeredTools")], + [ + "resource", + manifest.resources.map((item) => item.uri), + Reflect.get(server, "_registeredResources"), + ], + [ + "prompt", + manifest.prompts.map((item) => item.name), + Reflect.get(server, "_registeredPrompts"), + ], + ] as const; + for (const [kind, names, existing] of registeredCapabilities) { + for (const name of names) { + if (Object.hasOwn(existing, name)) { + return yield* new McpPassthroughUnavailableError({ + reason: `Client ${kind} collides with a built-in registration: ${name}`, + }); + } + } + } + yield* Effect.sync(() => { + for (const item of manifest.tools) { + const inputSchema = item.inputSchema + ? z.fromJSONSchema(item.inputSchema as Parameters[0]) + : {}; + server.registerTool( + item.name, + { description: item.description, inputSchema }, + async (args) => { + const result = await invokeClient({ + method: "tools/call", + params: { name: item.name, arguments: args }, + }); + return Effect.runPromiseWith(clientContext)(normalizeClientResult(result)); + }, + ); + } + for (const item of manifest.resources) { + server.registerResource( + item.name ?? item.uri, + item.uri, + { + description: item.description, + mimeType: item.mimeType, + }, + async (uri) => + Effect.runPromiseWith(clientContext)( + normalizeClientResourceResult( + await invokeClient({ method: "resources/read", params: { uri: uri.href } }), + uri.href, + ), + ), + ); + } + for (const item of manifest.prompts) { + server.registerPrompt( + item.name, + { + description: item.description, + argsSchema: Object.fromEntries( + (item.arguments ?? []).map((argument) => [ + argument.name, + z.string().describe(argument.description ?? ""), + ]), + ), + }, + async (args) => + Effect.runPromiseWith(clientContext)( + normalizeClientPromptResult( + await invokeClient({ + method: "prompts/get", + params: { name: item.name, arguments: args }, + }), + ), + ), + ); + } + }); + } + return server; }).pipe(Effect.withSpan("mcp.host.create_executor_server")); From 75aa2433d96589e50dbaa3a320b15ae0b99e8524 Mon Sep 17 00:00:00 2001 From: FRFlo <62435410+FRFlo@users.noreply.github.com> Date: Mon, 19 Oct 2026 12:34:11 +0200 Subject: [PATCH 2/2] fix(e2e): isolate Google OAuth rollout from CI flags --- e2e/scenarios/first-party-oauth.test.ts | 12 +++++------- e2e/setup/cloud.boot.ts | 4 ++++ 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/e2e/scenarios/first-party-oauth.test.ts b/e2e/scenarios/first-party-oauth.test.ts index bee0c78daa..db7d8ece4e 100644 --- a/e2e/scenarios/first-party-oauth.test.ts +++ b/e2e/scenarios/first-party-oauth.test.ts @@ -187,13 +187,11 @@ scenario( }); }); - // The Executor-owned Google app is withheld from every listing: it is no - // longer offered for new connections, so connecting Google means bringing - // your own OAuth app. It stays fully resolvable by slug, which the - // `oauth.start` cases below exercise — that is the guarantee for everyone - // already connected through it. Its reviewed consumer scope bundle, no - // longer introspectable from here, is asserted on the config it is built - // from, in apps/cloud/src/engine/first-party-oauth-clients.test.ts. + // The cloud E2E boot disables the Google rollout flag, so the app is + // withheld from listings here while remaining resolvable by slug. The + // `oauth.start` cases below prove existing connections can still use it. + // The reviewed consumer scope bundle is asserted on its config in + // apps/cloud/src/engine/first-party-oauth-clients.test.ts. const clients = yield* client.oauth.listClients(); expect( clients.find((candidate) => String(candidate.slug) === "first-party:google"), diff --git a/e2e/setup/cloud.boot.ts b/e2e/setup/cloud.boot.ts index 8314e7bc0b..e05a95b8ad 100644 --- a/e2e/setup/cloud.boot.ts +++ b/e2e/setup/cloud.boot.ts @@ -103,6 +103,10 @@ export const bootCloud = async (options: CloudBootOptions): Promise EXECUTOR_DIRECT_DATABASE_URL: "true", CLOUDFLARE_INCLUDE_PROCESS_ENV: "true", VITE_PUBLIC_SITE_URL: options.publicUrl, + // Keep the Google OAuth rollout disabled in E2E. CI may provide the + // production PostHog project key, but this scenario verifies the + // non-rollout path and must not depend on a user/org's flag assignment. + VITE_PUBLIC_POSTHOG_KEY: "", // The AuthKit domain (MCP OAuth metadata + JWKS) is the emulator too. MCP_AUTHKIT_DOMAIN: workosUrl, MCP_RESOURCE_ORIGIN: options.publicUrl,