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.