From d0f08f4db0c42009fe00989c0a12a6b48735c46d Mon Sep 17 00:00:00 2001 From: Tim Nunamaker Date: Wed, 14 Oct 2026 18:56:05 -0500 Subject: [PATCH] perf(auth): keep dispatch nonce out of event identity Resolved in-process requests carry a marked copy of the outer introspection context. Derive event token IDs from that context's grant ID, so deleting an aborted scope cannot make its nonce eligible for persistence. Ordinary requests keep their existing event identity. Assisted-by: AI Signed-off-by: Tim Nunamaker --- reference-implementation/server/index.ts | 44 +- .../server/internal-dispatch.ts | 140 +++++ .../server/request-helpers.ts | 5 +- .../server/routes/rs-hosted-mcp.ts | 96 ++- .../server/routes/rs-read.ts | 8 +- reference-implementation/server/transport.ts | 40 +- .../test/hosted-mcp-oauth.test.ts | 589 +++++++++++++++++- .../test/internal-dispatch.test.ts | 211 +++++++ .../test/rs-hosted-mcp-internal-base.test.ts | 8 +- .../vendor/mcp-server/test/rs-client.test.ts | 27 + 10 files changed, 1116 insertions(+), 52 deletions(-) create mode 100644 reference-implementation/server/internal-dispatch.ts create mode 100644 reference-implementation/test/internal-dispatch.test.ts diff --git a/reference-implementation/server/index.ts b/reference-implementation/server/index.ts index 4eddaa5a3..dfea65b3c 100644 --- a/reference-implementation/server/index.ts +++ b/reference-implementation/server/index.ts @@ -9,6 +9,7 @@ */ import { createHash, randomBytes } from "node:crypto"; import { existsSync, readFileSync, renameSync } from "node:fs"; +import { Socket } from "node:net"; import os from "node:os"; import path from "node:path"; import { isMainModule } from "../scripts/is-main-module.js"; @@ -227,6 +228,12 @@ import { createRemoteIntrospector, type IntrospectionCallerCredentials, } from "./introspection-http.ts"; +import { + getDispatchEntry, + getEventTokenId, + getRequestAbortSignal, + markDispatchedTokenInfo, +} from "./internal-dispatch.ts"; import { buildAuthorizationServerMetadata, buildClientEventSubscriptionsCapability, @@ -1801,8 +1808,7 @@ async function emitQueryRejected(context: QueryContext, req: ReqLike, err: ApiEr if (Object.hasOwn(context, "sourceDescriptor")) { data.source = context.sourceDescriptor ?? null; } - const authHeader = req.headers?.authorization; - const tokenId = typeof authHeader === "string" ? authHeader.slice(7) : null; + const tokenId = getEventTokenId(req.headers?.authorization, context.tokenInfo); await emitSpineEvent({ actor_id: context.actorId ?? null, actor_type: context.actorType ?? null, @@ -1836,8 +1842,7 @@ async function emitQueryReceived(context: QueryContext, req: ReqLike) { if (Object.hasOwn(context, "sourceDescriptor")) { data.source = context.sourceDescriptor ?? null; } - const authHeader = req.headers?.authorization; - const tokenId = typeof authHeader === "string" ? authHeader.slice(7) : null; + const tokenId = getEventTokenId(req.headers?.authorization, context.tokenInfo); await emitSpineEvent({ actor_id: context.actorId ?? null, actor_type: context.actorType ?? null, @@ -1916,8 +1921,7 @@ async function emitStateEvent( status: string, data: Record = {} ) { - const authHeader = req.headers?.authorization; - const tokenId = typeof authHeader === "string" ? authHeader.slice(7) : null; + const tokenId = getEventTokenId(req.headers?.authorization, req.tokenInfo); await emitSpineEvent({ actor_id: context.actorId ?? null, actor_type: context.actorType ?? null, @@ -2007,8 +2011,7 @@ async function emitMutationEvent( status: string, data: Record = {} ) { - const authHeader = req.headers?.authorization; - const tokenId = typeof authHeader === "string" ? authHeader.slice(7) : null; + const tokenId = getEventTokenId(req.headers?.authorization, req.tokenInfo); await emitSpineEvent({ actor_id: context.actorId ?? null, actor_type: context.actorType ?? null, @@ -2072,13 +2075,31 @@ async function requireTokenWithIntrospection( next: () => void, introspectToken: (token: string) => Promise ) { + const requestSignal = getRequestAbortSignal(req.raw); + if (requestSignal?.aborted) { + return; + } const auth = req.headers?.authorization; - if (!auth || typeof auth !== "string" || !auth.startsWith("Bearer ")) { + const isInjectRequest = !(req.raw?.socket instanceof Socket); + const bearer = typeof auth === "string" && auth.startsWith("Bearer ") ? auth.slice(7) : undefined; + const dispatchEntry = isInjectRequest && bearer ? getDispatchEntry(bearer) : undefined; + if (!dispatchEntry && (!auth || typeof auth !== "string" || !auth.startsWith("Bearer "))) { setProtectedResourceMetadataChallenge(res); return pdppError(res, 401, "authentication_error", "Missing Bearer token"); } - const token = auth.slice(7); - const info = await introspectToken(token); + const info: TokenInfo = dispatchEntry + ? structuredClone(dispatchEntry.info as TokenInfo) + : await introspectToken((auth as string).slice(7)); + if (dispatchEntry) { + markDispatchedTokenInfo(info); + } + if (requestSignal?.aborted) { + return; + } + if (dispatchEntry && typeof info.exp === "number" && info.exp <= Math.floor(Date.now() / 1000)) { + info.active = false; + info.inactive_reason ??= "grant_expired"; + } if (info.active && info.pdpp_token_kind === "client" && req.params?.stream) { const requestedConnection = resolveRequestConnectionId(req.query ?? {}).connectionId; try { @@ -7474,6 +7495,7 @@ function buildRsApp(opts: ServerOpts = {}) { mountRsHostedMcp(app, { createPackageProfile, createRsClient, + dispatchApp: app, explicitResource, getGrantPackageAccess, handleStreamableHttpRequest, diff --git a/reference-implementation/server/internal-dispatch.ts b/reference-implementation/server/internal-dispatch.ts new file mode 100644 index 000000000..9bb086b5a --- /dev/null +++ b/reference-implementation/server/internal-dispatch.ts @@ -0,0 +1,140 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 + +import { randomBytes } from "node:crypto"; + +export interface DispatchInfo { + active?: boolean; + exp?: number | null; + [key: string]: unknown; +} + +export interface DispatchEntry { + readonly info: Readonly; + readonly openedAt: number; + readonly resource: string; +} + +interface InjectResult { + headers: Record; + payload: string; + statusCode: number; +} + +export interface DispatchApp { + inject: (options: { + headers: Record; + method: "DELETE" | "GET" | "HEAD" | "OPTIONS" | "PATCH" | "POST" | "PUT"; + payload?: Buffer; + url: string; + }) => Promise; +} + +const registry = new Map(); +const requestSignals = new WeakMap(); +const dispatchedTokenInfos = new WeakSet(); + +export function setRequestAbortSignal(request: object, signal: AbortSignal): void { + requestSignals.set(request, signal); +} + +export function getRequestAbortSignal(request: object | undefined): AbortSignal | undefined { + return request ? requestSignals.get(request) : undefined; +} + +export function getDispatchRegistrySize(): number { + return registry.size; +} + +export function getDispatchEntry(nonce: string): DispatchEntry | undefined { + return registry.get(nonce); +} + +export function markDispatchedTokenInfo(info: object): void { + dispatchedTokenInfos.add(info); +} + +export function getEventTokenId(authorization: unknown, tokenInfo?: object | null): string | null { + if (tokenInfo && dispatchedTokenInfos.has(tokenInfo)) { + const grantId = (tokenInfo as { grant_id?: unknown }).grant_id; + return typeof grantId === "string" ? grantId : null; + } + if (typeof authorization !== "string" || !authorization.startsWith("Bearer ")) { + return null; + } + return authorization.slice(7); +} + +function createInProcessFetch(app: DispatchApp, nonce: string): typeof globalThis.fetch { + const dispatchFetch = async (input: string | URL | Request, init?: RequestInit) => { + const request = input instanceof Request ? input : new Request(input, init); + const url = new URL(request.url); + const headers: Record = {}; + request.headers.forEach((value, name) => { + if (name.toLowerCase() !== "authorization") { + headers[name] = value; + } + }); + headers.authorization = `Bearer ${nonce}`; + const body = ["GET", "HEAD"].includes(request.method) + ? undefined + : Buffer.from(await request.arrayBuffer()); + const response = await app.inject({ + headers, + method: request.method as + | "DELETE" + | "GET" + | "HEAD" + | "OPTIONS" + | "PATCH" + | "POST" + | "PUT", + ...(body === undefined ? {} : { payload: body }), + url: `${url.pathname}${url.search}`, + }); + const responseBody = [204, 205, 304].includes(response.statusCode) + ? null + : response.payload; + const responseHeaders = new Headers(); + for (const [name, value] of Object.entries(response.headers)) { + if (Array.isArray(value)) { + for (const item of value) { + responseHeaders.append(name, item); + } + } else if (value !== undefined) { + responseHeaders.set(name, value); + } + } + return new Response(responseBody, { + headers: responseHeaders, + status: response.statusCode, + }); + }; + return dispatchFetch; +} + +export async function runWithDispatchScope( + info: DispatchInfo, + resource: string, + app: DispatchApp, + fn: (fetch: typeof globalThis.fetch) => Promise, + signal?: AbortSignal, +): Promise { + if (signal?.aborted) { + throw new DOMException("The operation was aborted", "AbortError"); + } + const nonce = randomBytes(16).toString("hex"); + registry.set(nonce, { + info: structuredClone(info), + openedAt: Date.now(), + resource, + }); + const removeOnAbort = () => registry.delete(nonce); + signal?.addEventListener("abort", removeOnAbort, { once: true }); + try { + return await fn(createInProcessFetch(app, nonce)); + } finally { + signal?.removeEventListener("abort", removeOnAbort); + registry.delete(nonce); + } +} diff --git a/reference-implementation/server/request-helpers.ts b/reference-implementation/server/request-helpers.ts index a77fd266a..719bfddea 100644 --- a/reference-implementation/server/request-helpers.ts +++ b/reference-implementation/server/request-helpers.ts @@ -12,6 +12,7 @@ */ import { emitSpineEvent, generateSpineId, type SpineEventInput } from "../lib/spine.ts"; +import { getEventTokenId } from "./internal-dispatch.ts"; import { codeToStatus, typeFor } from "./routes/ref-error-status.ts"; // ─── Structural request/response shapes ────────────────────────────────────── @@ -158,7 +159,7 @@ export async function emitQueryRejected( stream_id: context.streamId || null, subject_id: context.tokenInfo?.subject_id || null, subject_type: "subject", - token_id: req.headers.authorization?.slice(7) || null, + token_id: getEventTokenId(req.headers.authorization, context.tokenInfo), trace_id: context.traceId, // Cast bridges `exactOptionalPropertyTypes`: several context fields are // `string | null | undefined` at runtime (e.g. `scenarioId` from @@ -199,7 +200,7 @@ export async function emitQueryReceived(context: QueryContext, req: RequestLike) stream_id: context.streamId || null, subject_id: context.tokenInfo?.subject_id || null, subject_type: "subject", - token_id: req.headers.authorization?.slice(7) || null, + token_id: getEventTokenId(req.headers.authorization, context.tokenInfo), trace_id: context.traceId, // See the cast note in `emitQueryRejected` above — same runtime-vs- // exactOptional bridge; no value coercion. diff --git a/reference-implementation/server/routes/rs-hosted-mcp.ts b/reference-implementation/server/routes/rs-hosted-mcp.ts index 332c24447..bd3d02c79 100644 --- a/reference-implementation/server/routes/rs-hosted-mcp.ts +++ b/reference-implementation/server/routes/rs-hosted-mcp.ts @@ -20,6 +20,7 @@ // without `.d.ts` declarations, so they cannot be imported directly from a // strict-mode `.ts` file. +import { getRequestAbortSignal, type DispatchApp, runWithDispatchScope } from "../internal-dispatch.ts"; import { isTrustedMetadataRequestOrigin, protectedResourceMetadataUrlForResource, @@ -38,7 +39,13 @@ interface RouteRequest extends ResolvePublicUrlRequest { readonly headers: Readonly>; readonly method: string; readonly path?: string; - readonly raw?: { readonly url?: string }; + readonly raw?: { + readonly aborted?: boolean; + readonly socket?: { readonly destroyed?: boolean; readonly remoteAddress?: string }; + readonly url?: string; + on?: (event: string, listener: () => void) => void; + off?: (event: string, listener: () => void) => void; + }; readonly tokenInfo?: { readonly grant_package_id?: string; readonly pdpp_token_kind?: string; @@ -49,6 +56,12 @@ interface RouteRequest extends ResolvePublicUrlRequest { interface RouteResponse { end: () => void; locals: Record; + raw?: { + destroyed?: boolean; + off?: (event: string, listener: () => void) => void; + on?: (event: string, listener: () => void) => void; + writableEnded?: boolean; + }; send: (body: unknown) => void; setHeader: (name: string, value: string) => void; status: (code: number) => RouteResponse; @@ -100,14 +113,18 @@ export interface MountRsHostedMcpContext { fetch: typeof globalThis.fetch; }) => unknown; /** - * `createRsClient` from `./package-rs-client.js` — builds one single-bearer - * RsClient against a chosen fetch base. Injected for the same JS-only reason. - * Used for the standalone (`client`-token) path so its self-calls can use the - * internal RS base too (parity with the package path's child clients). + * `createRsClient` from `./package-rs-client.js` — builds one RsClient over + * the supplied fetch. The standalone path uses the outer bearer on the + * client; the in-process fetch replaces it with the live dispatch nonce. */ - createRsClient: (options: { providerUrl: string; accessToken: string; fetch: typeof globalThis.fetch }) => unknown; + createRsClient: (options: { + providerUrl: string; + accessToken: string; + fetch: typeof globalThis.fetch; + }) => unknown; /** Resolved RS public URL (or null; adapter derives it per-request). */ readonly explicitResource: string | null | undefined; + readonly dispatchApp: DispatchApp; /** From auth.js: resolve active grant-package members for a package token. */ getGrantPackageAccess: (packageId: string) => Promise; /** @@ -197,12 +214,6 @@ function buildMcpWebRequest(req: RouteRequest, resource: string): Request { return new Request(url.toString(), init); } -function extractInboundMcpToken(req: RouteRequest): string { - const authHeader = req.headers.authorization; - const authValue = Array.isArray(authHeader) ? (authHeader[0] ?? "") : (authHeader ?? ""); - return authValue.slice(7); -} - interface SendWebResponseOptions { readonly iconLink?: string; } @@ -318,20 +329,21 @@ export function mountRsHostedMcp(app: AppLike, ctx: MountRsHostedMcpContext): vo } function buildStandaloneMcpServerOptions( + req: RouteRequest, resource: string, internalBase: string, - accessToken: string + dispatchFetch: typeof globalThis.fetch ): McpServerOptions { - // Standalone (`client`-token) path: build the single-bearer RsClient - // against the internal base so its self-calls avoid the public-edge - // hairpin too — parity with the package path's child clients. The - // advertised `providerUrl` - // stays the public `resource` (display/provenance only; all fetches go - // through the injected rsClient). When no internal base is configured, - // internalBase === resource, so this is a no-op vs prior behavior. + // Standalone (`client`-token) path: dispatch through the real RS route + // chain in-process. The `providerUrl` stays public for display/provenance; + // all resource reads use the injected dispatch fetch. + const authorization = req.headers.authorization; + if (typeof authorization !== "string" || !authorization.startsWith("Bearer ")) { + throw new TypeError("standalone hosted MCP requires the outer bearer token"); + } const rsClient = ctx.createRsClient({ - accessToken, - fetch: globalThis.fetch, + accessToken: authorization.slice(7), + fetch: dispatchFetch, providerUrl: internalBase, }); return { @@ -350,15 +362,19 @@ export function mountRsHostedMcp(app: AppLike, ctx: MountRsHostedMcpContext): vo res: RouteResponse, resource: string, internalBase: string, - inboundToken: string + dispatchFetch: typeof globalThis.fetch ): Promise { if (req.tokenInfo?.pdpp_token_kind === "mcp_package") { return buildPackageMcpServerOptions(req, res, resource, internalBase); } - return buildStandaloneMcpServerOptions(resource, internalBase, inboundToken); + return buildStandaloneMcpServerOptions(req, resource, internalBase, dispatchFetch); } async function handleHostedMcp(req: RouteRequest, res: RouteResponse): Promise { + const requestSignal = getRequestAbortSignal(req.raw); + if (requestSignal?.aborted || req.raw?.aborted || req.raw?.socket?.destroyed || res.raw?.destroyed) { + return; + } // Advertised identity (what clients discover and call) — always public. const resource = resolvePublicUrl(req, explicitResource); // Server-internal fetch base for the adapter's own child-grant self-calls. @@ -367,14 +383,36 @@ export function mountRsHostedMcp(app: AppLike, ctx: MountRsHostedMcpContext): vo // The internal base is fetch-only: it is never advertised, never written // into discovery metadata, and never carried in issued-token audiences. const internalBase = internalResource ?? resource; - const inboundToken = extractInboundMcpToken(req); - - const mcpServerOptions = await buildHostedMcpServerOptions(req, res, resource, internalBase, inboundToken); + const webRequest = buildMcpWebRequest(req, resource); + if (req.tokenInfo?.pdpp_token_kind === "client") { + const response = await runWithDispatchScope( + req.tokenInfo, + `${resource}/mcp`, + ctx.dispatchApp, + async (dispatchFetch) => { + const mcpServerOptions = await buildHostedMcpServerOptions( + req, + res, + resource, + internalBase, + dispatchFetch + ); + if (!mcpServerOptions) { + return null; + } + return ctx.handleStreamableHttpRequest(webRequest, mcpServerOptions); + }, + requestSignal + ); + if (response && !requestSignal?.aborted && !res.raw?.destroyed) { + await sendWebResponse(res, response, { iconLink: hostedMcpIconLink(resource) }); + } + return; + } + const mcpServerOptions = await buildHostedMcpServerOptions(req, res, resource, internalBase, globalThis.fetch); if (!mcpServerOptions) { return; } - - const webRequest = buildMcpWebRequest(req, resource); const response = await ctx.handleStreamableHttpRequest(webRequest, mcpServerOptions); await sendWebResponse(res, response, { iconLink: hostedMcpIconLink(resource) }); } diff --git a/reference-implementation/server/routes/rs-read.ts b/reference-implementation/server/routes/rs-read.ts index 9a89055f2..bf60b3c26 100644 --- a/reference-implementation/server/routes/rs-read.ts +++ b/reference-implementation/server/routes/rs-read.ts @@ -33,6 +33,7 @@ import { BlobsReadNotFoundError, executeBlobsRead, } from "../../operations/rs-blobs-read/index.ts"; +import { getEventTokenId } from "../internal-dispatch.ts"; import { type ConnectorsListDependencies, type ConnectorsListInput, @@ -400,9 +401,8 @@ export interface MountRsReadContext { ) => void; } -function authorizationTokenId(req: RouteRequest): string | null { - const auth = req.headers.authorization; - return typeof auth === "string" ? auth.slice(7) : null; +function authorizationTokenId(req: RouteRequest, tokenInfo: TokenInfo): string | null { + return getEventTokenId(req.headers.authorization, tokenInfo); } // Resolve the request-time `connection_id`, honoring the deprecated @@ -615,7 +615,7 @@ function emitDisclosureServed( status: "succeeded", subject_id: tokenInfo.subject_id || null, subject_type: "subject", - token_id: authorizationTokenId(args.req), + token_id: authorizationTokenId(args.req, tokenInfo), trace_id: args.traceId, }; if (args.streamId !== undefined) { diff --git a/reference-implementation/server/transport.ts b/reference-implementation/server/transport.ts index 8f8f4ec7d..6a73c496e 100644 --- a/reference-implementation/server/transport.ts +++ b/reference-implementation/server/transport.ts @@ -62,6 +62,7 @@ import { isResponseCanary, } from "./contract-validation.ts"; import { HOSTED_INGEST_MAX_REQUEST_BYTES } from "./hosted-ingest-limits.ts"; +import { setRequestAbortSignal } from "./internal-dispatch.ts"; // Header name the reference sets on responses to expose the protocol trace // ID (handler-set via setReferenceTraceId in server/index.js). @@ -162,6 +163,7 @@ function loadQsParser(): QsParser { const localRequire = createRequire(import.meta.url); const qs = loadQsParser(); +const requestAbortCleanup = new WeakMap void>(); function canResolvePrettyTransport(): boolean { try { @@ -314,7 +316,11 @@ function buildRouteSchema(manifest: RouteManifest): Record | * request-completion log lines. We emit our own completion record in an * `onResponse` hook so it can carry PDPP's `trace_id` alongside `req_id`. */ -function buildFastify({ loggerInstance }: { loggerInstance: FastifyBaseLogger }): FastifyInstance { +function buildFastify({ + loggerInstance, +}: { + loggerInstance: FastifyBaseLogger; +}): FastifyInstance { const fastify = Fastify({ bodyLimit: HOSTED_INGEST_MAX_REQUEST_BYTES, // match previous express.text() limit disableRequestLogging: true, @@ -359,7 +365,23 @@ function buildFastify({ loggerInstance }: { loggerInstance: FastifyBaseLogger }) // Force `Connection: close` on every response. See the note above the // Fastify config block. An `onRequest` hook sets it before handlers run so // there's no race with streaming replies. - fastify.addHook("onRequest", (_request, reply, done) => { + fastify.addHook("onRequest", (request, reply, done) => { + const controller = new AbortController(); + const abortOnRequest = () => controller.abort(); + const abortOnResponseClose = () => { + if (!reply.raw.writableEnded) controller.abort(); + }; + if (request.raw.aborted || request.raw.socket.destroyed || reply.raw.destroyed) { + controller.abort(); + } else { + request.raw.once("aborted", abortOnRequest); + reply.raw.once("close", abortOnResponseClose); + } + setRequestAbortSignal(request.raw, controller.signal); + requestAbortCleanup.set(request, () => { + request.raw.off("aborted", abortOnRequest); + reply.raw.off("close", abortOnResponseClose); + }); reply.header("connection", "close"); done(); }); @@ -395,6 +417,8 @@ function buildFastify({ loggerInstance }: { loggerInstance: FastifyBaseLogger }) // record the spec promises, not a per-status-code shape. // biome-ignore lint/suspicious/useAwait: The async signature is part of this caller-facing contract. fastify.addHook("onResponse", async (request, reply) => { + requestAbortCleanup.get(request)?.(); + requestAbortCleanup.delete(request); const traceId = reply.getHeader?.(PDPP_TRACE_ID_HEADER) || request.__pdppTraceId; request.log.info( { @@ -1215,6 +1239,7 @@ export function createApp({ logger, __requestValidationAllowlistForTest }: Creat // transport-coverage test can assert every contract manifest has its // binding declared at registration time. Object.assign(fastify.server, { + __pdppInject: app.inject, __pdppFastify: fastify, __pdppRegisteredRoutes: [...registeredRoutes], }); @@ -1226,6 +1251,17 @@ export function createApp({ logger, __requestValidationAllowlistForTest }: Creat const app = { delete: del, + inject: (options: { + headers: Record; + method: "DELETE" | "GET" | "HEAD" | "OPTIONS" | "PATCH" | "POST" | "PUT"; + payload?: Buffer; + url: string; + }) => + fastify.inject({ + ...options, + headers: options.headers, + }), + // Escape hatch — tests and runtime adapters may need the raw Fastify // instance or its underlying http.Server. fastify, diff --git a/reference-implementation/test/hosted-mcp-oauth.test.ts b/reference-implementation/test/hosted-mcp-oauth.test.ts index d3523fc85..e24c251ce 100644 --- a/reference-implementation/test/hosted-mcp-oauth.test.ts +++ b/reference-implementation/test/hosted-mcp-oauth.test.ts @@ -7,15 +7,17 @@ import assert from "node:assert/strict"; import { createHash, generateKeyPairSync, randomBytes, sign } from "node:crypto"; import { mkdtempSync, readFileSync, rmSync } from "node:fs"; import http from "node:http"; +import { Writable } from "node:stream"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; import test from "node:test"; import { fileURLToPath } from "node:url"; -import Database from "./helpers/sqlite-driver.ts"; +import pino from "pino"; import { buildPendingConsentRequestUri, countGrantPackagesForOwner, getGrantPackageAccess, + introspect, revokeGrant, revokeGrantPackage, } from "../server/auth.ts"; @@ -23,16 +25,22 @@ import { canonicalConnectorKey, canonicalConnectorKeyFromManifest } from "../ser import { closeDb, getDb, initDb } from "../server/db.ts"; import { encodeHostedMcpSelection, encodeHostedMcpStreamSelection, hostedMcpSourceKey } from "../server/hosted-mcp-selection.ts"; import { scopeFieldsInputName, scopeSinceInputName, scopeUntilInputName } from "../server/hosted-mcp-stream-scope.ts"; -import { computeHostedMcpDecisionDigest } from "../server/routes/as-consent-ui-helpers.ts"; import { startServer } from "../server/index.ts"; +import { + getDispatchRegistrySize, + type DispatchApp, + runWithDispatchScope, +} from "../server/internal-dispatch.ts"; import { basicIntrospectionAuthorization } from "../server/introspection-http.ts"; import { ingestRecord, queryRecordsAcrossBindings, resolveReadRequestBindings } from "../server/records.ts"; +import { computeHostedMcpDecisionDigest } from "../server/routes/as-consent-ui-helpers.ts"; import { createSqliteConnectorInstanceStore } from "../server/stores/connector-instance-store.ts"; import { bindHostedMcpConsentChallenge } from "./helpers/hosted-mcp-consent-challenge.ts"; import { TEST_INTROSPECTION_SERVER_OPTS, TEST_RS_INTROSPECTION_CREDENTIALS, } from "./helpers/introspection-test-credentials.ts"; +import Database from "./helpers/sqlite-driver.ts"; const __dirname = dirname(fileURLToPath(import.meta.url)); const REFERENCE_IMPL_DIR = join(__dirname, ".."); @@ -1203,6 +1211,583 @@ test("hosted MCP OAuth code flow issues a scoped client token usable at /mcp", a } }); +test("hosted MCP client schema call introspects its bearer once", async () => { + let introspectionCount = 0; + const innerNetworkCalls: string[] = []; + const logLines: string[] = []; + let nonceSocketProbe = false; + const logger = pino( + { level: "trace" }, + new Writable({ + write(chunk, _encoding, callback) { + logLines.push(String(chunk)); + callback(); + }, + }) + ); + const originalFetch = globalThis.fetch; + const server = await startServer({ + asPort: 0, + dbPath: ":memory:", + introspectionFetch: async (input, init) => { + introspectionCount += 1; + return fetch(input, init); + }, + logger, + ownerAuthPassword: "", + quiet: true, + rsPort: 0, + ...TEST_INTROSPECTION_SERVER_OPTS, + }); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + type InjectOptions = { headers?: Record; method: string; payload?: unknown; url: string }; + const fastify = ( + server.rsServer as unknown as { __pdppFastify: { inject: (options: InjectOptions) => Promise } } + ).__pdppFastify; + const originalInject = fastify.inject.bind(fastify); + const dispatchNonces: string[] = []; + const innerAuthorizationHeaders: Array = []; + let liveSocketNonceStatus: number | undefined; + fastify.inject = async (options) => { + if (options.url.startsWith("/v1/")) { + const authorization = options.headers?.authorization; + const nonce = authorization?.startsWith("Bearer ") ? authorization.slice(7) : ""; + dispatchNonces.push(nonce); + innerAuthorizationHeaders.push(options.headers?.authorization); + if (nonce && liveSocketNonceStatus === undefined) { + nonceSocketProbe = true; + try { + liveSocketNonceStatus = ( + await fetchJson(`${rsUrl}/v1/schema`, { headers: { Authorization: `Bearer ${nonce}` } }) + ).status; + } finally { + nonceSocketProbe = false; + } + } + } + return originalInject(options); + }; + globalThis.fetch = async (input, init) => { + const url = new URL(input instanceof Request ? input.url : input.toString()); + if (url.pathname.startsWith("/v1/") && !nonceSocketProbe) { + innerNetworkCalls.push(`${url.pathname}${url.search}`); + } + return originalFetch(input, init); + }; + + try { + const manifest = await registerAuthorizedSpotify(asUrl); + const client = await registerAuthCodeClient(asUrl); + const { accessToken } = await completeOauthCodeFlow({ asUrl, client, manifest }); + introspectionCount = 0; + + const schema = await postMcpJson(rsUrl, accessToken, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }); + + assert.equal(schema.status, 200); + assert.equal(resultOf(schema).isError, undefined); + assert.equal(introspectionCount, 2, "the outer bearer and live-nonce socket probe each introspect once"); + assert.equal(liveSocketNonceStatus, 401, "a real socket cannot use the live inject nonce as its bearer"); + assert.deepEqual(innerNetworkCalls, [], "inner /v1 work must not open a loopback HTTP request"); + assert.ok(dispatchNonces.length > 0, "the real /v1 route ran through Fastify inject"); + assert.ok(dispatchNonces.every((nonce) => /^[0-9a-f]{32}$/.test(nonce))); + assert.ok( + innerAuthorizationHeaders.every((authorization) => authorization === `Bearer ${dispatchNonces[0]}`), + "the inner bearer is the live nonce, never the outer user's bearer", + ); + assert.ok(dispatchNonces.every((nonce) => nonce !== accessToken)); + const firstNonce = mustExist(dispatchNonces[0], "the dispatch nonce was captured"); + for (const nonce of dispatchNonces) { + assert.ok(!JSON.stringify(schema.body).includes(nonce), "nonce is absent from response bodies"); + } + const replayedNonce = await fetchJson(`${rsUrl}/v1/schema`, { + headers: { Authorization: `Bearer ${firstNonce}` }, + }); + assert.equal(replayedNonce.status, 401, "the nonce is rejected after the dispatch scope closes"); + logger.flush?.(); + assert.ok(logLines.every((line) => !dispatchNonces.some((nonce) => line.includes(nonce)))); + const eventRows = getDb().prepare("SELECT token_id, data_json FROM spine_events").all(); + assert.ok(eventRows.length > 0, "the hosted read emitted spine events"); + const nonceEventRows = eventRows.filter((row) => JSON.stringify(row).includes(firstNonce)); + assert.equal( + nonceEventRows.length, + 0, + `nonce is absent from event data (fields: ${nonceEventRows.map((row) => + `token_id=${row.token_id === firstNonce}, data_json=${String(row.data_json).includes(firstNonce)}` + ).join(", ")})`, + ); + } finally { + globalThis.fetch = originalFetch; + fastify.inject = originalInject; + await closeServer(server); + } +}); + +test("dispatch identity never persists its nonce after abort, handler failure, or completion", async () => { + const logLines: string[] = []; + const logger = pino( + { level: "trace" }, + new Writable({ + write(chunk, _encoding, callback) { + logLines.push(String(chunk)); + callback(); + }, + }) + ); + const server = await startServer({ + asPort: 0, + dbPath: ":memory:", + logger, + ownerAuthPassword: "", + quiet: true, + rsPort: 0, + ...TEST_INTROSPECTION_SERVER_OPTS, + }); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + const fastify = ( + server.rsServer as unknown as { __pdppFastify: { inject: (options: unknown) => Promise } } + ).__pdppFastify; + const statementPrototype = Object.getPrototypeOf(getDb().prepare("SELECT 1")) as { + run: (...args: unknown[]) => unknown; + }; + const originalRun = statementPrototype.run; + + try { + const manifest = await registerAuthorizedSpotify(asUrl); + const client = await registerAuthCodeClient(asUrl); + const { accessToken } = await completeOauthCodeFlow({ asUrl, client, manifest }); + const tokenInfo = await introspect(accessToken); + assert.equal(tokenInfo.active, true); + + const runScenario = async (mode: "abort" | "throw" | "complete") => { + getDb().exec("DELETE FROM spine_events"); + const controller = new AbortController(); + let nonce = ""; + let abortAtFirstEventInsert = mode !== "complete"; + statementPrototype.run = function (...args: unknown[]) { + if ( + abortAtFirstEventInsert && + /INSERT\s+INTO\s+spine_events/i.test(String((this as { source?: unknown }).source)) + ) { + abortAtFirstEventInsert = false; + controller.abort(); + } + return originalRun.apply(this, args); + }; + + const dispatchApp: DispatchApp = { + inject: async (options) => { + nonce = mustExist(options.headers.authorization?.slice(7), "internal inject carries a nonce bearer"); + return fastify.inject(options) as ReturnType; + }, + }; + let result: { body: string; headers: Record; status: number } | undefined; + let thrown: unknown; + try { + result = await runWithDispatchScope( + tokenInfo, + rsUrl, + dispatchApp, + async (inProcessFetch) => { + const response = await inProcessFetch(`${rsUrl}/v1/schema`); + const body = await response.text(); + if (mode === "throw") { + throw new Error("simulated outer handler failure"); + } + return { body, headers: Object.fromEntries(response.headers.entries()), status: response.status }; + }, + controller.signal, + ); + } catch (error) { + thrown = error; + } + + logger.flush?.(); + const persistedEvents = getDb().prepare("SELECT * FROM spine_events").all(); + const eventAndTraceData = JSON.stringify(persistedEvents); + assert.ok(nonce, "the internal inject call exposed a nonce to the test harness"); + assert.ok(!eventAndTraceData.includes(nonce), `${mode}: persisted event and trace data excludes the nonce`); + assert.ok(!(JSON.stringify(result) ?? "").includes(nonce), `${mode}: response data excludes the nonce`); + assert.ok(logLines.every((line) => !line.includes(nonce)), `${mode}: logs exclude the nonce`); + assert.equal(getDispatchRegistrySize(), 0, `${mode}: dispatch scope is deleted`); + assert.ok(persistedEvents.some((event) => event.event_type === "disclosure.served")); + assert.ok(persistedEvents.every((event) => event.token_id === tokenInfo.grant_id)); + if (mode === "throw") { + assert.match(String(thrown), /simulated outer handler failure/); + } else { + assert.equal(thrown, undefined); + } + if (mode === "abort") { + assert.equal(result?.status, 200, "the admitted route completes after scope cleanup"); + } else if (mode === "complete") { + assert.equal(result?.status, 200); + } + }; + + await runScenario("throw"); + await runScenario("abort"); + await runScenario("complete"); + } finally { + statementPrototype.run = originalRun; + logger.flush?.(); + await closeServer(server); + } +}); + +test("disconnect during introspection opens no dispatch scope", async () => { + let introspectionCount = 0; + let blockIntrospection = false; + let markIntrospectionPaused: (() => void) | undefined; + let markIntrospectionFinished: (() => void) | undefined; + let releaseIntrospection: (() => void) | undefined; + const introspectionPaused = new Promise((resolve) => { + markIntrospectionPaused = resolve; + }); + const introspectionFinished = new Promise((resolve) => { + markIntrospectionFinished = resolve; + }); + const introspectionRelease = new Promise((resolve) => { + releaseIntrospection = resolve; + }); + const server = await startServer({ + asPort: 0, + dbPath: ":memory:", + introspectionFetch: async (input, init) => { + introspectionCount += 1; + const response = await fetch(input, init); + if (blockIntrospection) { + markIntrospectionPaused?.(); + await introspectionRelease; + markIntrospectionFinished?.(); + } + return response; + }, + ownerAuthPassword: "", + quiet: true, + rsPort: 0, + ...TEST_INTROSPECTION_SERVER_OPTS, + }); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + let markInnerStarted: (() => void) | undefined; + let releaseInner: (() => void) | undefined; + const innerStarted = new Promise((resolve) => { + markInnerStarted = resolve; + }); + const innerRelease = new Promise((resolve) => { + releaseInner = resolve; + }); + const fastify = ( + server.rsServer as unknown as { + __pdppFastify: { + inject: (options: { method: string; url: string }) => Promise; + }; + } + ).__pdppFastify; + const originalInject = fastify.inject.bind(fastify); + let innerRouteCount = 0; + fastify.inject = async (options) => { + if (options.url.startsWith("/v1/")) { + innerRouteCount += 1; + markInnerStarted?.(); + await innerRelease; + } + return originalInject(options); + }; + + try { + const manifest = await registerAuthorizedSpotify(asUrl); + const client = await registerAuthCodeClient(asUrl); + const { accessToken } = await completeOauthCodeFlow({ asUrl, client, manifest }); + introspectionCount = 0; + blockIntrospection = true; + + const payload = JSON.stringify({ + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }); + const controller = new AbortController(); + const request = fetch(`${rsUrl}/mcp`, { + body: payload, + headers: { + accept: "application/json, text/event-stream", + authorization: `Bearer ${accessToken}`, + "content-type": "application/json", + }, + method: "POST", + signal: controller.signal, + }).catch(() => undefined); + + await introspectionPaused; + controller.abort(); + await request; + await new Promise((resolve) => setTimeout(resolve, 75)); + releaseIntrospection?.(); + await introspectionFinished; + await Promise.race([innerStarted, new Promise((resolve) => setTimeout(resolve, 100))]); + + assert.equal(introspectionCount, 1); + assert.equal(getDispatchRegistrySize(), 0, "disconnect before auth completes must not open a context"); + assert.equal(innerRouteCount, 0, "a disconnected request must not dispatch resource reads"); + } finally { + releaseIntrospection?.(); + releaseInner?.(); + fastify.inject = originalInject; + await closeServer(server); + } +}); + +test("expanded hosted MCP query_records introspects its bearer once", async () => { + let introspectionCount = 0; + const server = await startServer({ + asPort: 0, + dbPath: ":memory:", + introspectionFetch: async (input, init) => { + introspectionCount += 1; + return fetch(input, init); + }, + ownerAuthPassword: "", + quiet: true, + rsPort: 0, + ...TEST_INTROSPECTION_SERVER_OPTS, + }); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + type FastifyInject = (options: { headers?: Record; method: string; payload?: unknown; url: string }) => Promise; + const fastify = ( + server.rsServer as unknown as { __pdppFastify: { inject: FastifyInject } } + ).__pdppFastify; + const originalInject = fastify.inject.bind(fastify); + let innerRouteCount = 0; + const innerAuthorizationHeaders: Array = []; + fastify.inject = async (options) => { + if (options.url.startsWith("/v1/")) { + innerRouteCount += 1; + innerAuthorizationHeaders.push(options.headers?.authorization); + } + return originalInject(options); + }; + + try { + const manifest = await registerAuthorizedSpotify(asUrl); + const client = await registerAuthCodeClient(asUrl); + const { accessToken } = await completeOauthCodeFlow({ asUrl, client, manifest }); + introspectionCount = 0; + + const expandedRead = await postMcpJson(rsUrl, accessToken, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { + arguments: { expand: ["fixture_relation"], stream: manifest.streams[0]?.name }, + name: "query_records", + }, + }); + + assert.equal(expandedRead.status, 200); + assert.equal(resultOf(expandedRead).isError, true, "client-token expansions are rejected by the v0.1 query contract"); + assert.equal(introspectionCount, 1, "schema preflight and record reads share the outer context"); + assert.equal(innerRouteCount, 2, "the rejected expansion performs schema preflight and one record request"); + const innerNonces = innerAuthorizationHeaders.map((authorization) => { + const header = mustExist(authorization, "an internal request must carry the live nonce bearer"); + assert.ok(header.startsWith("Bearer ")); + return header.slice(7); + }); + assert.equal(new Set(innerNonces).size, 1, "all inner requests use this outer dispatch nonce"); + const innerNonce = mustExist(innerNonces[0], "internal request nonce must exist"); + assert.notEqual(innerNonce, accessToken, "the outer user bearer is not forwarded"); + assert.ok(!JSON.stringify(expandedRead).includes(innerNonce)); + } finally { + fastify.inject = originalInject; + await closeServer(server); + } +}); + +test("hosted MCP re-introspects after grant revocation between outer requests", async () => { + let introspectionCount = 0; + const server = await startServer({ + asPort: 0, + dbPath: ":memory:", + introspectionFetch: async (input, init) => { + introspectionCount += 1; + return fetch(input, init); + }, + ownerAuthPassword: "", + quiet: true, + rsPort: 0, + ...TEST_INTROSPECTION_SERVER_OPTS, + }); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + + try { + const manifest = await registerAuthorizedSpotify(asUrl); + const client = await registerAuthCodeClient(asUrl); + const { accessToken, grantId } = await completeOauthCodeFlow({ asUrl, client, manifest }); + introspectionCount = 0; + + const first = await postMcpJson(rsUrl, accessToken, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }); + assert.equal(first.status, 200); + assert.equal(introspectionCount, 1); + + await revokeGrant(mustExist(grantId, "authorization flow issues grant"), { + request_id: "mcp-inprocess-revocation-test", + }); + const second = await postMcpJson(rsUrl, accessToken, { + id: 2, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }); + + assert.equal(second.status, 401); + assert.equal(introspectionCount, 2, "the next outer request must introspect again"); + } finally { + await closeServer(server); + } +}); + +test("external requests cannot forge inject authorization with nonce-shaped tokens or legacy headers", async () => { + const server = await startOpenTestServer(); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + + try { + const spotify = await registerAuthorizedSpotify(asUrl); + const github = await registerAuthorizedGithub(asUrl); + const clientA = await registerAuthCodeClient(asUrl); + const clientB = await registerAuthCodeClient(asUrl); + const grantA = await completeOauthCodeFlow({ asUrl, client: clientA, manifest: spotify }); + const grantB = await completeOauthCodeFlow({ asUrl, client: clientB, manifest: github }); + const inProcessInject = (server.rsServer as unknown as { __pdppInject: DispatchApp["inject"] }) + .__pdppInject; + assert.equal( + (await inProcessInject({ headers: { authorization: `Bearer ${grantA.accessToken}` }, method: "GET", url: "/v1/schema" })) + .statusCode, + 200 + ); + + const forgedInject = await inProcessInject({ + headers: { authorization: `Bearer ${randomBytes(16).toString("hex")}` }, + method: "GET", + url: "/v1/schema", + }); + assert.equal(forgedInject.statusCode, 401, "an inject request cannot use a nonce outside a live scope"); + const forgedSocket = await fetchJson(`${rsUrl}/v1/schema`, { + headers: { Authorization: `Bearer ${randomBytes(16).toString("hex")}` }, + }); + assert.equal(forgedSocket.status, 401, "a real socket always uses normal token introspection"); + + const legacyHeaders = await fetchJson(`${rsUrl}/v1/schema`, { + headers: { + "PdPp-InTeRnAl-DiSpAtCh": randomBytes(16).toString("hex"), + "X-PdPp-InTeRnAl-InJeCt": "01234567-89ab-cdef-0123-456789abcdef", + }, + }); + assert.equal(legacyHeaders.status, 401, "legacy headers cannot authorize a socket request"); + + const duplicateHeaders = await new Promise((resolve, reject) => { + const request = http.request( + `${rsUrl}/v1/schema`, + { + headers: [ + "Host", + new URL(rsUrl).host, + "authorization", + `Bearer ${randomBytes(16).toString("hex")}`, + "AUTHORIZATION", + `Bearer ${randomBytes(16).toString("hex")}`, + "x-pdpp-internal-inject", + "01234567-89ab-cdef-0123-456789abcdef", + "X-PDPP-INTERNAL-INJECT", + "01234567-89ab-cdef-0123-456789abcdef", + ], + method: "GET", + }, + (response) => { + response.resume(); + response.on("end", () => resolve(response.statusCode ?? 0)); + } + ); + request.on("error", reject); + request.end(); + }); + assert.equal(duplicateHeaders, 401, "duplicate mixed-case authorization headers fail closed"); + + const otherGrant = await fetchJson(`${rsUrl}/v1/schema`, { + headers: { Authorization: `Bearer ${grantB.accessToken}` }, + }); + assert.equal(otherGrant.status, 200); + const otherConnectorIds = new Set( + schemaStreamRows((otherGrant.body.data ?? otherGrant.body) as Record) + .map((stream) => stream.source.connector_id) + .filter((connectorId): connectorId is string => typeof connectorId === "string") + ); + assert.ok(otherConnectorIds.has(github.connector_id)); + assert.ok(!otherConnectorIds.has(spotify.connector_id)); + } finally { + await closeServer(server); + } +}); + +test("concurrent hosted MCP requests keep each grant context isolated", async () => { + const server = await startOpenTestServer(); + const asUrl = `http://localhost:${server.asPort}`; + const rsUrl = `http://localhost:${server.rsPort}`; + + try { + const spotify = await registerAuthorizedSpotify(asUrl); + const github = await registerAuthorizedGithub(asUrl); + const [clientA, clientB] = await Promise.all([registerAuthCodeClient(asUrl), registerAuthCodeClient(asUrl)]); + const [grantA, grantB] = await Promise.all([ + completeOauthCodeFlow({ asUrl, client: clientA, manifest: spotify }), + completeOauthCodeFlow({ asUrl, client: clientB, manifest: github }), + ]); + const [responseA, responseB] = await Promise.all([ + postMcpJson(rsUrl, grantA.accessToken, { + id: 1, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }), + postMcpJson(rsUrl, grantB.accessToken, { + id: 2, + jsonrpc: "2.0", + method: "tools/call", + params: { arguments: {}, name: "schema" }, + }), + ]); + + assert.equal(responseA.status, 200); + assert.equal(responseB.status, 200); + const connectorsA = new Set( + schemaStreamRows(structuredContentData(resultOf(responseA))).map((stream) => stream.source.connector_id) + ); + const connectorsB = new Set( + schemaStreamRows(structuredContentData(resultOf(responseB))).map((stream) => stream.source.connector_id) + ); + assert.ok(connectorsA.has(spotify.connector_id)); + assert.ok(!connectorsA.has(github.connector_id)); + assert.ok(connectorsB.has(github.connector_id)); + assert.ok(!connectorsB.has(spotify.connector_id)); + } finally { + await closeServer(server); + } +}); + test("OAuth token lifetime and refresh eligibility follow the persisted grant contract", async () => { const server = await startOpenTestServer(); const asUrl = `http://localhost:${server.asPort}`; diff --git a/reference-implementation/test/internal-dispatch.test.ts b/reference-implementation/test/internal-dispatch.test.ts new file mode 100644 index 000000000..0e7307d1c --- /dev/null +++ b/reference-implementation/test/internal-dispatch.test.ts @@ -0,0 +1,211 @@ +// Copyright The PDP-Connect Contributors +// SPDX-License-Identifier: Apache-2.0 + +import assert from "node:assert/strict"; +import test from "node:test"; +import { + getDispatchEntry, + getDispatchRegistrySize, + runWithDispatchScope, +} from "../server/internal-dispatch.ts"; + +const tokenInfo = { + active: true, + aud: "https://resource.example/mcp", + exp: Math.floor(Date.now() / 1000) + 60, + grant_id: "grant_dispatch_test", + pdpp_token_kind: "client", + scope: { streams: ["orders"] }, +}; + +test("dispatch scope injects with its live nonce and removes it after success", async () => { + let nonce = ""; + let authorization: string | undefined; + let injectedUrl = ""; + const app = { + inject: async (options: { + headers: Record; + method: string; + payload?: Buffer; + url: string; + }) => { + authorization = options.headers.authorization; + nonce = authorization?.startsWith("Bearer ") ? authorization.slice(7) : ""; + const storedInfo = getDispatchEntry(nonce)?.info; + assert.notEqual(storedInfo, tokenInfo, "scope stores a structured clone"); + assert.notEqual( + storedInfo?.scope, + tokenInfo.scope, + "nested authorization state is cloned too", + ); + assert.equal("pdpp-internal-dispatch" in options.headers, false); + assert.equal(authorization, `Bearer ${nonce}`); + injectedUrl = options.url; + return { + headers: { "content-type": "application/json" }, + payload: '{"ok":true}', + statusCode: 200, + }; + }, + }; + + const response = await runWithDispatchScope( + tokenInfo, + "https://resource.example/mcp", + app, + (dispatchFetch: typeof fetch) => + dispatchFetch( + new Request("https://resource.example/v1/schema?view=compact", { + headers: { + Authorization: "Bearer must-not-forward", + "X-Trace": "trace-1", + }, + }), + ), + ); + + assert.match(nonce, /^[0-9a-f]{32}$/); + assert.equal(authorization, `Bearer ${nonce}`); + assert.equal(injectedUrl, "/v1/schema?view=compact"); + assert.equal(response.status, 200); + const body = await response.text(); + assert.equal(body, '{"ok":true}'); + assert.equal( + body.includes(nonce), + false, + "nonce is absent from the response body", + ); + assert.ok( + [...response.headers].every(([name, value]) => !name.includes(nonce) && !value.includes(nonce)), + "nonce is absent from response headers", + ); + assert.equal(getDispatchRegistrySize(), 0); + assert.equal( + getDispatchEntry(nonce), + undefined, + "nonce cannot be replayed after scope completion", + ); +}); + +test("dispatch scope removes its nonce after a tool failure", async () => { + const app = { + inject: async () => { + throw new Error("tool failed"); + }, + }; + + await assert.rejects( + runWithDispatchScope( + tokenInfo, + "https://resource.example/mcp", + app, + async () => { + throw new Error("tool failed"); + }, + ), + /tool failed/, + ); + assert.equal(getDispatchRegistrySize(), 0); +}); + +test("concurrent dispatch scopes keep distinct authorization contexts", async () => { + const observed = new Map(); + const app = { + inject: async ({ headers }: { headers: Record }) => { + const authorization = headers.authorization; + const nonce = authorization?.startsWith("Bearer ") ? authorization.slice(7) : undefined; + observed.set(nonce ?? "", getDispatchEntry(nonce ?? "")?.info.grant_id as string); + await new Promise((resolve) => setTimeout(resolve, 5)); + return { headers: {}, payload: "{}", statusCode: 200 }; + }, + }; + + await Promise.all( + ["grant_a", "grant_b"].map((grant_id) => + runWithDispatchScope( + { ...tokenInfo, grant_id }, + "https://resource.example/mcp", + app, + (dispatchFetch: typeof fetch) => + dispatchFetch(new Request("https://resource.example/v1/schema")), + ), + ), + ); + + assert.equal(observed.size, 2); + assert.ok( + [...observed.keys()].every((nonce) => /^[0-9a-f]{32}$/.test(nonce)), + ); + assert.equal(getDispatchRegistrySize(), 0); +}); + +test("P3 TODO: network requests reject MCP-audience bearers at the /v1 mount", { + todo: "P3 per-mount audience enforcement is not in this change", +}, async () => { + let innerAuthorization: string | undefined; + const app = { + inject: async ({ headers }: { headers: Record }) => { + innerAuthorization = headers.authorization; + return { headers: {}, payload: "{}", statusCode: 200 }; + }, + }; + await runWithDispatchScope( + tokenInfo, + "https://resource.example/mcp", + app, + (dispatchFetch: typeof fetch) => + dispatchFetch( + new Request("https://resource.example/v1/schema", { + headers: { Authorization: "Bearer mcp-audience-token" }, + }), + ), + ); + assert.equal( + innerAuthorization?.startsWith("Bearer "), + true, + "the internal request uses a bearer nonce rather than the MCP bearer", + ); + assert.notEqual(innerAuthorization, "Bearer mcp-audience-token"); +}); + +test("aborting the outer request deletes its dispatch context immediately", async () => { + const controller = new AbortController(); + let finish: (() => void) | undefined; + const pending = runWithDispatchScope( + tokenInfo, + "https://resource.example/mcp", + { inject: async () => ({ headers: {}, payload: "{}", statusCode: 200 }) }, + () => + new Promise((resolve) => { + finish = resolve; + }), + controller.signal, + ); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(getDispatchRegistrySize(), 1); + controller.abort(); + assert.equal(getDispatchRegistrySize(), 0); + finish?.(); + await pending; +}); + +test("an already-aborted request cannot open a dispatch scope", async () => { + const controller = new AbortController(); + controller.abort(); + let dispatched = false; + + await assert.rejects( + runWithDispatchScope( + tokenInfo, + "https://resource.example/mcp", + { inject: async () => ({ headers: {}, payload: "{}", statusCode: 200 }) }, + async () => { + dispatched = true; + }, + controller.signal, + ), + { name: "AbortError" }, + ); + assert.equal(dispatched, false, "an aborted outer request must not begin tool work"); + assert.equal(getDispatchRegistrySize(), 0); +}); diff --git a/reference-implementation/test/rs-hosted-mcp-internal-base.test.ts b/reference-implementation/test/rs-hosted-mcp-internal-base.test.ts index 03a43f187..5aebbc6a8 100644 --- a/reference-implementation/test/rs-hosted-mcp-internal-base.test.ts +++ b/reference-implementation/test/rs-hosted-mcp-internal-base.test.ts @@ -155,6 +155,11 @@ function makeContext({ internalResource }: { internalResource: string | null | u seen.singleGrantProviderUrl = providerUrl; return { __fakeRsClient: true }; }, + dispatchApp: { + async inject() { + return { headers: {}, payload: "{}", statusCode: 200 }; + }, + }, explicitResource: PUBLIC_RESOURCE, // biome-ignore lint/suspicious/useAwait: Async callback preserves the dependency contract and rejection timing. async getGrantPackageAccess() { @@ -238,8 +243,7 @@ test("F1 wiring fallback: with no internal base configured, child self-calls fal test("F1 wiring (single-grant): client-token self-calls use the internal base; advertised stays public", async () => { const seen = await driveHandler({ internalResource: INTERNAL_BASE, makeRequest: makeClientRequest }); - // The single-bearer RsClient's fetch base is the INTERNAL base (the - // single-grant extension — parity with the package path). + // The single-grant RsClient uses the INTERNAL base for its provider URL. assert.equal( seen.singleGrantProviderUrl, INTERNAL_BASE, diff --git a/reference-implementation/vendor/mcp-server/test/rs-client.test.ts b/reference-implementation/vendor/mcp-server/test/rs-client.test.ts index 2e1c7f1d3..e0061878b 100644 --- a/reference-implementation/vendor/mcp-server/test/rs-client.test.ts +++ b/reference-implementation/vendor/mcp-server/test/rs-client.test.ts @@ -2,6 +2,7 @@ // SPDX-License-Identifier: Apache-2.0 import assert from "node:assert/strict"; +import { createServer } from "node:http"; import { test } from "node:test"; import { type QueryParams, RsClient } from "../src/rs-client.ts"; @@ -52,6 +53,32 @@ test("attaches bearer token and forwards query params", async () => { assert.equal(headers?.Authorization, "Bearer scoped-abc"); }); +test("requires a bearer before a custom network fetch can be used", async () => { + let echoRequests = 0; + const echoServer = createServer((request, response) => { + echoRequests += 1; + response.setHeader("content-type", "application/json"); + response.end(JSON.stringify({ authorization: request.headers.authorization ?? null })); + }); + await new Promise((resolve) => echoServer.listen(0, "127.0.0.1", resolve)); + const address = echoServer.address(); + assert.ok(address && typeof address === "object"); + const fetch = (input: string | URL | Request, init?: RequestInit) => globalThis.fetch(input, init); + + try { + assert.throws( + () => new RsClient({ + fetch, + providerUrl: `http://127.0.0.1:${address.port}`, + } as ConstructorParameters[0]), + /RsClient requires accessToken/, + ); + assert.equal(echoRequests, 0, "a tokenless RsClient cannot make an echo request"); + } finally { + await new Promise((resolve, reject) => echoServer.close((error) => (error ? reject(error) : resolve()))); + } +}); + test("rejects object-valued query params instead of JSON-stringifying them", async () => { let called = false; // biome-ignore lint/suspicious/useAwait: async required to satisfy the Promise-returning fetch/getJson contract this fixture implements; a synchronous return type is not assignable to the caller's injected dependency.