perf(auth): authorize each MCP call once via in-process dispatch

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<string, unknown> = {}
 ) {
-  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<string, unknown> = {}
 ) {
-  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<TokenInfo>
 ) {
+  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,

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<DispatchInfo>;
+	readonly openedAt: number;
+	readonly resource: string;
+}
+
+interface InjectResult {
+	headers: Record<string, string | string[] | undefined>;
+	payload: string;
+	statusCode: number;
+}
+
+export interface DispatchApp {
+	inject: (options: {
+		headers: Record<string, string>;
+		method: "DELETE" | "GET" | "HEAD" | "OPTIONS" | "PATCH" | "POST" | "PUT";
+		payload?: Buffer;
+		url: string;
+	}) => Promise<InjectResult>;
+}
+
+const registry = new Map<string, DispatchEntry>();
+const requestSignals = new WeakMap<object, AbortSignal>();
+const dispatchedTokenInfos = new WeakSet<object>();
+
+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<string, string> = {};
+		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<Result>(
+	info: DispatchInfo,
+	resource: string,
+	app: DispatchApp,
+	fn: (fetch: typeof globalThis.fetch) => Promise<Result>,
+	signal?: AbortSignal,
+): Promise<Result> {
+	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);
+	}
+}

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.