diff --git a/.changeset/rfc-9421-auto-sign-client-transports.md b/.changeset/rfc-9421-auto-sign-client-transports.md new file mode 100644 index 000000000..c0017803a --- /dev/null +++ b/.changeset/rfc-9421-auto-sign-client-transports.md @@ -0,0 +1,80 @@ +--- +'@adcp/client': minor +--- + +Auto-apply RFC 9421 request signing to outbound MCP and A2A calls inside +`ProtocolClient` / `AdCPClient`. Follow-up to the signing primitives shipped +previously: the library now wires the signer into `StreamableHTTPClientTransport` +and the A2A `fetchImpl` automatically when an `AgentConfig.request_signing` +block is present. + +Behavior: + +- On first outbound call for an agent with `request_signing`, the client + fetches `get_adcp_capabilities` (unsigned — the discovery op is exempt) and + caches the seller's `request_signing` capability per-agent with a 300s TTL. +- Subsequent calls consult the cache to decide per-operation whether to + sign — precedence matches the spec: buyer `always_sign` > seller + `required_for` > seller `warn_for` (shadow-mode telemetry) > seller + `supported_for` (buyer opted in via `sign_supported`). +- Content-digest coverage honors the seller's `covers_content_digest` policy + (`required` / `forbidden` / `either`) per-request. +- Transport connection caches disambiguate by a per-key fingerprint (hash of + `kid` + private scalar) so two tenants that misconfigure the same `kid` + but hold distinct private keys cannot collide on a shared cached + transport and sign each other's traffic. +- `get_adcp_capabilities` and MCP/A2A protocol-layer RPCs (`initialize`, + `tools/list`, A2A card discovery) always pass through unsigned. +- OAuth-gated agents with signing: `callMCPToolWithOAuth` threads the + signing context through to the transport fetch, so OAuth flows don't + silently drop signatures. +- Priming failures fail open with a 60s negative cache: a transient seller + discovery outage doesn't wedge every subsequent call. `always_sign` ops + still get signed with sensible content-digest defaults; ops the seller + might have listed in `required_for` reach the wire unsigned and are + rejected visibly with `request_signature_required`, which retries re-prime. +- Concurrent cold-cache fans-out share one `get_adcp_capabilities` fetch + via an in-flight pending-map stored on the `CapabilityCache` instance + itself — so two tenants with separate `CapabilityCache` instances get + independent in-flight tables, and embedders who construct their own + cache don't race against the default cache. +- `AgentSigningContext.invalidate()` evicts this context's capability + entry so callers don't have to rebuild the cache key from the agent's + identifying fields when they want to force a re-prime. +- Signing-reserved headers (`Signature`, `Signature-Input`, `Content-Digest`) + supplied by a caller's `customHeaders` are scrubbed before the signer + runs — a misconfigured header cannot silently break or bypass the RFC + 9421 signature output. +- `extractAdcpOperation` throws on unsupported body shapes (Blob, FormData, + ReadableStream) rather than silently passing the request unsigned — the + seller's `required_for` contract is not broken by SDK body-format drift. + +New field on `AgentConfig`: `request_signing?: AgentRequestSigningConfig` +(kid, alg, `AdcpPrivateJsonWebKey` with required `d`, agent_url, optional +`always_sign[]` and `sign_supported`). + +New sub-barrels: + +- `@adcp/client/signing/client` — signer, canonicalization, fetch wrapper, + capability cache, and the auto-wiring helpers a buyer building an + AdCPClient needs. +- `@adcp/client/signing/server` — verifier pipeline, Express-shaped + middleware, JWKS / replay / revocation stores, error taxonomy. + +The existing `@adcp/client/signing` barrel continues to export the union of +both sub-barrels, so existing consumers keep working. New code should +import from whichever half matches its role — coding agents reading a file +cold only need to hold one side of the taxonomy. + +New exports on `@adcp/client/signing/client`: `CapabilityCache`, +`buildCapabilityCacheKey`, `defaultCapabilityCache`, +`buildAgentSigningContext`, `buildAgentSigningFetch`, +`ensureCapabilityLoaded`, `extractAdcpOperation`, `shouldSignOperation`, +`resolveCoverContentDigest`, `toSignerKey`, `CAPABILITY_OP`, +`CoverContentDigestPredicate`. `AgentSigningContext` gains an +`invalidate()` method. `CachedCapability` gains an optional `staleAt` +deadline for negative-cache entries. + +`createSigningFetch` now accepts `coverContentDigest` as either `boolean` +or `(url, init) => boolean` so the seller policy can be resolved per +request without rebuilding the wrapper. diff --git a/package.json b/package.json index a691fd02e..1d2cecedd 100644 --- a/package.json +++ b/package.json @@ -40,6 +40,16 @@ "import": "./dist/lib/signing/index.js", "require": "./dist/lib/signing/index.js", "types": "./dist/lib/signing/index.d.ts" + }, + "./signing/client": { + "import": "./dist/lib/signing/client.js", + "require": "./dist/lib/signing/client.js", + "types": "./dist/lib/signing/client.d.ts" + }, + "./signing/server": { + "import": "./dist/lib/signing/server.js", + "require": "./dist/lib/signing/server.js", + "types": "./dist/lib/signing/server.d.ts" } }, "typesVersions": { @@ -61,6 +71,12 @@ ], "signing": [ "dist/lib/signing/index.d.ts" + ], + "signing/client": [ + "dist/lib/signing/client.d.ts" + ], + "signing/server": [ + "dist/lib/signing/server.d.ts" ] } }, diff --git a/src/lib/protocols/a2a.ts b/src/lib/protocols/a2a.ts index ae692e2bd..a7e71724c 100644 --- a/src/lib/protocols/a2a.ts +++ b/src/lib/protocols/a2a.ts @@ -10,6 +10,7 @@ import { AuthenticationRequiredError, is401Error } from '../errors'; import { discoverOAuthMetadata } from '../auth/oauth/discovery'; import { withSpan, injectTraceHeaders } from '../observability/tracing'; import { isAgentCardPath, buildCardUrls } from '../utils/a2a-discovery'; +import { buildAgentSigningFetch, type AgentSigningContext } from '../signing/client'; if (!A2AClient) { throw new Error('A2A SDK client is required. Please install @a2a-js/sdk'); @@ -40,13 +41,13 @@ const callContextStorage = new AsyncLocalStorage(); const a2aClientCache = new Map>(); const pendingA2AClients = new Map>>(); -function a2aCacheKey(agentUrl: string, authToken?: string): string { - if (!authToken) return agentUrl; +function a2aCacheKey(agentUrl: string, authToken?: string, signingCacheKey?: string): string { // 64-bit hash prefix — cache key disambiguator, not a security boundary. // The cached client closes over the full authToken; a hypothetical hash // collision still sends the original token, not the colliding one. - const tokenHash = createHash('sha256').update(authToken).digest('hex').slice(0, 16); - return `${agentUrl}::${tokenHash}`; + const tokenSuffix = authToken ? `::${createHash('sha256').update(authToken).digest('hex').slice(0, 16)}` : ''; + const signingSuffix = signingCacheKey ? `::${signingCacheKey}` : ''; + return `${agentUrl}${tokenSuffix}${signingSuffix}`; } /** @@ -61,16 +62,17 @@ export function closeA2AConnections(): void { async function getOrCreateA2AClient( agentUrl: string, - authToken: string | undefined + authToken: string | undefined, + signingContext: AgentSigningContext | undefined ): Promise> { - const cacheKey = a2aCacheKey(agentUrl, authToken); + const cacheKey = a2aCacheKey(agentUrl, authToken, signingContext?.cacheKey); const cached = a2aClientCache.get(cacheKey); if (cached) return cached; const pending = pendingA2AClients.get(cacheKey); if (pending) return pending; - const promise = createA2AClient(agentUrl, authToken) + const promise = createA2AClient(agentUrl, authToken, signingContext) .then(client => { a2aClientCache.set(cacheKey, client); return client; @@ -85,9 +87,10 @@ async function getOrCreateA2AClient( async function createA2AClient( agentUrl: string, - authToken: string | undefined + authToken: string | undefined, + signingContext: AgentSigningContext | undefined ): Promise> { - const fetchImpl = buildFetchImpl(authToken); + const fetchImpl = buildFetchImpl(authToken, signingContext); const cardUrls = buildCardUrls(agentUrl); const context = callContextStorage.getStore(); @@ -113,8 +116,12 @@ async function createA2AClient( return client; } -function buildFetchImpl(authToken: string | undefined) { - return async (url: string | URL | Request, options?: RequestInit) => { +function buildFetchImpl(authToken: string | undefined, signingContext: AgentSigningContext | undefined) { + // Inner fetch handles auth/header injection and 401 detection. If the + // agent has request-signing configured, we wrap it with the AdCP signing + // fetch so the signature covers the exact bytes we're about to send (auth + // headers included, since the signer re-reads the final header record). + const baseFetch = async (url: string | URL | Request, options?: RequestInit) => { const context = callContextStorage.getStore(); const existingHeaders: Record = {}; @@ -170,6 +177,20 @@ function buildFetchImpl(authToken: string | undefined) { return response; }; + + if (!signingContext) return baseFetch; + + // The signing wrapper assembles headers into the signature base. We invoke + // it first so the signer sees the caller-supplied headers; baseFetch then + // overlays auth/trace headers afterwards — A2A's auth scheme (bearer) is + // not among the MANDATORY_COMPONENTS and is injected by the counterparty's + // transport layer, not signed. + const signingFetch = buildAgentSigningFetch({ + upstream: (input, init) => baseFetch(input as any, init), + signing: signingContext.signing, + getCapability: signingContext.getCapability, + }); + return signingFetch; } export async function callA2ATool( @@ -179,7 +200,8 @@ export async function callA2ATool( authToken?: string, debugLogs: DebugLogEntry[] = [], pushNotificationConfig?: PushNotificationConfig, - customHeaders?: Record + customHeaders?: Record, + signingContext?: AgentSigningContext ): Promise { return withSpan( 'adcp.a2a.call_tool', @@ -194,7 +216,16 @@ export async function callA2ATool( got401Ref: { value: false }, }; return callContextStorage.run(context, () => - callA2AToolImpl(agentUrl, toolName, parameters, authToken, debugLogs, pushNotificationConfig, context) + callA2AToolImpl( + agentUrl, + toolName, + parameters, + authToken, + debugLogs, + pushNotificationConfig, + context, + signingContext + ) ); } ); @@ -207,10 +238,11 @@ async function callA2AToolImpl( authToken: string | undefined, debugLogs: DebugLogEntry[], pushNotificationConfig: PushNotificationConfig | undefined, - context: A2ACallContext + context: A2ACallContext, + signingContext: AgentSigningContext | undefined ): Promise { try { - const client = await getOrCreateA2AClient(agentUrl, authToken); + const client = await getOrCreateA2AClient(agentUrl, authToken, signingContext); const requestPayload: { message: { @@ -281,7 +313,7 @@ async function callA2AToolImpl( } catch (error: unknown) { if (is401Error(error, context.got401Ref.value)) { // Evict this cache entry — token may have expired or been revoked. - a2aClientCache.delete(a2aCacheKey(agentUrl, authToken)); + a2aClientCache.delete(a2aCacheKey(agentUrl, authToken, signingContext?.cacheKey)); debugLogs.push({ type: 'error', diff --git a/src/lib/protocols/index.ts b/src/lib/protocols/index.ts index ac67750c1..972087888 100644 --- a/src/lib/protocols/index.ts +++ b/src/lib/protocols/index.ts @@ -37,6 +37,7 @@ import { createNonInteractiveOAuthProvider } from '../auth/oauth'; import { validateAgentUrl } from '../validation'; import { withSpan } from '../observability/tracing'; import { ADCP_MAJOR_VERSION } from '../version'; +import { buildAgentSigningContext, CAPABILITY_OP, ensureCapabilityLoaded } from '../signing/client'; /** * Universal protocol client - automatically routes to the correct protocol implementation @@ -79,6 +80,27 @@ export class ProtocolClient { const authToken = getAuthToken(agent); + // RFC 9421 signing context. Built once per call; the transport + // attaches a fetch wrapper that reads the cached capability on every + // outbound request. `get_adcp_capabilities` is exempt from signing + // (it's the discovery call itself) and also triggers cache priming + // for any other op on agents with `request_signing` configured. + const signingContext = buildAgentSigningContext(agent); + if (signingContext && toolName !== CAPABILITY_OP) { + await ensureCapabilityLoaded(agent, signingContext, primeArgs => + ProtocolClient.callTool( + agent, + CAPABILITY_OP, + primeArgs, + debugLogs, + undefined, + undefined, + undefined, + serverVersion + ) + ); + } + // Declare AdCP major version on every request so sellers can validate compatibility. // Skip for v2 servers — they don't recognise the field and strict-schema agents reject it. const argsWithVersion = serverVersion === 'v2' ? args : { adcp_major_version: ADCP_MAJOR_VERSION, ...args }; @@ -116,12 +138,21 @@ export class ProtocolClient { authProvider, debugLogs, customHeaders: agent.headers, + signingContext, }); } // Use callMCPToolWithTasks which auto-detects server tasks capability // and falls back to standard callTool when tasks are not supported - return callMCPToolWithTasks(agent.agent_uri, toolName, argsWithWebhook, authToken, debugLogs, agent.headers); + return callMCPToolWithTasks( + agent.agent_uri, + toolName, + argsWithWebhook, + authToken, + debugLogs, + agent.headers, + signingContext ? { signingContext } : undefined + ); } else if (agent.protocol === 'a2a') { // For A2A, pass pushNotificationConfig separately (not in skill parameters) return callA2ATool( @@ -131,7 +162,8 @@ export class ProtocolClient { authToken, debugLogs, pushNotificationConfig, - agent.headers + agent.headers, + signingContext ); } else { throw new Error(`Unsupported protocol: ${agent.protocol}`); diff --git a/src/lib/protocols/mcp-tasks.ts b/src/lib/protocols/mcp-tasks.ts index 54260c11b..312efa9f3 100644 --- a/src/lib/protocols/mcp-tasks.ts +++ b/src/lib/protocols/mcp-tasks.ts @@ -15,6 +15,7 @@ import type { TaskInfo } from '../core/ConversationTypes'; import { withCachedConnection } from './mcp'; import { createMCPAuthHeaders } from '../auth'; import { withSpan, injectTraceHeaders } from '../observability/tracing'; +import type { AgentSigningContext } from '../signing/client'; /** Response shape returned by MCPClient.callTool(). */ type CallToolResponse = { @@ -145,7 +146,7 @@ export async function callMCPToolWithTasks( authToken?: string, debugLogs: DebugLogEntry[] = [], customHeaders?: Record, - options?: { workingTimeout?: number } + options?: { workingTimeout?: number; signingContext?: AgentSigningContext } ): Promise { return withSpan( 'adcp.mcp.call_tool', @@ -181,116 +182,149 @@ export async function callMCPToolWithTasks( }); } - return withCachedConnection(agentUrl, authToken, authHeaders, debugLogs, toolName, async client => { - // Check if server supports MCP Tasks - if (!serverSupportsTasks(client)) { + return withCachedConnection( + agentUrl, + authToken, + authHeaders, + debugLogs, + toolName, + async client => { + // Check if server supports MCP Tasks + if (!serverSupportsTasks(client)) { + debugLogs.push({ + type: 'info', + message: `MCP Tasks: Server does not support tasks, using standard callTool for ${toolName}`, + timestamp: new Date().toISOString(), + }); + const response = (await client.callTool({ name: toolName, arguments: args })) as CallToolResponse; + + debugLogs.push({ + type: response?.isError ? 'error' : 'success', + message: `MCP: Tool ${toolName} response received (${response?.isError ? 'error' : 'success'})`, + timestamp: new Date().toISOString(), + response: response, + }); + + return response; + } + + // Ensure tool metadata is cached so the SDK's isToolTask() works correctly. + // Without this, callToolStream silently skips task creation for tools that + // declare taskSupport: 'optional' | 'required'. + await ensureToolsListed(client); + debugLogs.push({ type: 'info', - message: `MCP Tasks: Server does not support tasks, using standard callTool for ${toolName}`, + message: `MCP Tasks: Server supports tasks, using callToolStream for ${toolName}`, timestamp: new Date().toISOString(), }); - const response = (await client.callTool({ name: toolName, arguments: args })) as CallToolResponse; - debugLogs.push({ - type: response?.isError ? 'error' : 'success', - message: `MCP: Tool ${toolName} response received (${response?.isError ? 'error' : 'success'})`, - timestamp: new Date().toISOString(), - response: response, + // Use callToolStream which handles the full task lifecycle + const stream = client.experimental.tasks.callToolStream({ name: toolName, arguments: args }, undefined, { + timeout: workingTimeout, + resetTimeoutOnProgress: true, }); - return response; - } - - // Ensure tool metadata is cached so the SDK's isToolTask() works correctly. - // Without this, callToolStream silently skips task creation for tools that - // declare taskSupport: 'optional' | 'required'. - await ensureToolsListed(client); - - debugLogs.push({ - type: 'info', - message: `MCP Tasks: Server supports tasks, using callToolStream for ${toolName}`, - timestamp: new Date().toISOString(), - }); - - // Use callToolStream which handles the full task lifecycle - const stream = client.experimental.tasks.callToolStream({ name: toolName, arguments: args }, undefined, { - timeout: workingTimeout, - resetTimeoutOnProgress: true, - }); - - let capturedTaskId: string | undefined; - let capturedTask: { taskId: string; status: string; pollInterval?: number } | undefined; - - try { - for await (const message of stream) { - switch (message.type) { - case 'taskCreated': - capturedTaskId = message.task.taskId; - capturedTask = message.task; - debugLogs.push({ - type: 'info', - message: `MCP Tasks: Task created ${capturedTaskId} for ${toolName}`, - timestamp: new Date().toISOString(), - }); - break; - - case 'taskStatus': - capturedTask = message.task; - debugLogs.push({ - type: 'info', - message: `MCP Tasks: Status update for ${capturedTaskId}: ${message.task.status}`, - timestamp: new Date().toISOString(), - }); - break; - - case 'result': { - debugLogs.push({ - type: 'success', - message: `MCP: Tool ${toolName} response received (success)`, - timestamp: new Date().toISOString(), - response: message.result, - }); - return message.result as CallToolResponse; - } + let capturedTaskId: string | undefined; + let capturedTask: { taskId: string; status: string; pollInterval?: number } | undefined; + + try { + for await (const message of stream) { + switch (message.type) { + case 'taskCreated': + capturedTaskId = message.task.taskId; + capturedTask = message.task; + debugLogs.push({ + type: 'info', + message: `MCP Tasks: Task created ${capturedTaskId} for ${toolName}`, + timestamp: new Date().toISOString(), + }); + break; + + case 'taskStatus': + capturedTask = message.task; + debugLogs.push({ + type: 'info', + message: `MCP Tasks: Status update for ${capturedTaskId}: ${message.task.status}`, + timestamp: new Date().toISOString(), + }); + break; + + case 'result': { + debugLogs.push({ + type: 'success', + message: `MCP: Tool ${toolName} response received (success)`, + timestamp: new Date().toISOString(), + response: message.result, + }); + return message.result as CallToolResponse; + } - case 'error': { - debugLogs.push({ - type: 'error', - message: `MCP Tasks: Error for ${toolName}: ${message.error.message}`, - timestamp: new Date().toISOString(), - }); - // The MCP Tasks SDK error event may strip structured content. - // If we have a taskId, fetch the full result to recover adcp_error data - // and return it as a proper isError response for downstream unwrapping. - if (capturedTaskId) { - try { - const taskResult = await client.experimental.tasks.getTaskResult(capturedTaskId); - const content = taskResult?.content as Array<{ type: string; text?: string }> | undefined; - if (content) { - return { - isError: true, - content, - structuredContent: taskResult?.structuredContent, - } as unknown as CallToolResponse; + case 'error': { + debugLogs.push({ + type: 'error', + message: `MCP Tasks: Error for ${toolName}: ${message.error.message}`, + timestamp: new Date().toISOString(), + }); + // The MCP Tasks SDK error event may strip structured content. + // If we have a taskId, fetch the full result to recover adcp_error data + // and return it as a proper isError response for downstream unwrapping. + if (capturedTaskId) { + try { + const taskResult = await client.experimental.tasks.getTaskResult(capturedTaskId); + const content = taskResult?.content as Array<{ type: string; text?: string }> | undefined; + if (content) { + return { + isError: true, + content, + structuredContent: taskResult?.structuredContent, + } as unknown as CallToolResponse; + } + } catch { + // Failed to fetch task result — fall through to throw } - } catch { - // Failed to fetch task result — fall through to throw } + throw message.error; } - throw message.error; } } + } catch (error) { + // If we timed out but have a taskId, return a working status + // so the caller can poll via getMCPTaskStatus/getMCPTaskResult + if (capturedTaskId && error instanceof Error && error.message.includes('Timeout')) { + debugLogs.push({ + type: 'info', + message: `MCP Tasks: Timeout for ${toolName}, returning working status with taskId ${capturedTaskId}`, + timestamp: new Date().toISOString(), + }); + + return { + structuredContent: { + status: 'working', + task_id: capturedTaskId, + poll_interval: capturedTask?.pollInterval, + }, + }; + } + // Servers may return a synchronous tool error (e.g. VERSION_UNSUPPORTED) + // rather than creating a task. The Tasks SDK rejects these with a Zod + // validation error about a missing `task` field. When no task was + // captured, fall back to standard callTool so the error response + // reaches the caller intact. + const msg = error instanceof Error ? error.message : String(error); + if (!capturedTaskId && /Invalid task creation result/i.test(msg)) { + debugLogs.push({ + type: 'info', + message: `MCP Tasks: Server returned synchronous response for ${toolName}, falling back to callTool`, + timestamp: new Date().toISOString(), + }); + return (await client.callTool({ name: toolName, arguments: args })) as CallToolResponse; + } + throw error; } - } catch (error) { - // If we timed out but have a taskId, return a working status - // so the caller can poll via getMCPTaskStatus/getMCPTaskResult - if (capturedTaskId && error instanceof Error && error.message.includes('Timeout')) { - debugLogs.push({ - type: 'info', - message: `MCP Tasks: Timeout for ${toolName}, returning working status with taskId ${capturedTaskId}`, - timestamp: new Date().toISOString(), - }); + // Stream ended without result — shouldn't happen with well-behaved servers + if (capturedTaskId) { return { structuredContent: { status: 'working', @@ -299,36 +333,11 @@ export async function callMCPToolWithTasks( }, }; } - // Servers may return a synchronous tool error (e.g. VERSION_UNSUPPORTED) - // rather than creating a task. The Tasks SDK rejects these with a Zod - // validation error about a missing `task` field. When no task was - // captured, fall back to standard callTool so the error response - // reaches the caller intact. - const msg = error instanceof Error ? error.message : String(error); - if (!capturedTaskId && /Invalid task creation result/i.test(msg)) { - debugLogs.push({ - type: 'info', - message: `MCP Tasks: Server returned synchronous response for ${toolName}, falling back to callTool`, - timestamp: new Date().toISOString(), - }); - return (await client.callTool({ name: toolName, arguments: args })) as CallToolResponse; - } - throw error; - } - - // Stream ended without result — shouldn't happen with well-behaved servers - if (capturedTaskId) { - return { - structuredContent: { - status: 'working', - task_id: capturedTaskId, - poll_interval: capturedTask?.pollInterval, - }, - }; - } - - throw new Error(`MCP Tasks: callToolStream for ${toolName} ended without result or task`); - }); + + throw new Error(`MCP Tasks: callToolStream for ${toolName} ended without result or task`); + }, + options?.signingContext + ); } ); } diff --git a/src/lib/protocols/mcp.ts b/src/lib/protocols/mcp.ts index 01b1b6367..24662cc6d 100644 --- a/src/lib/protocols/mcp.ts +++ b/src/lib/protocols/mcp.ts @@ -13,6 +13,7 @@ import { createMCPAuthHeaders } from '../auth'; import { is401Error } from '../errors'; import type { DebugLogEntry } from '../types/adcp'; import { withSpan, injectTraceHeaders } from '../observability/tracing'; +import { buildAgentSigningFetch, type AgentSigningContext } from '../signing/client'; // Re-export for convenience export { UnauthorizedError }; @@ -66,10 +67,11 @@ function trackStreamableHTTPUrl(url: string): void { } } -function connectionCacheKey(agentUrl: string, authToken?: string): string { - if (!authToken) return agentUrl; - const tokenHash = createHash('sha256').update(authToken).digest('hex').slice(0, 16); - return `${agentUrl}::${tokenHash}`; +function connectionCacheKey(agentUrl: string, authToken?: string, signingCacheKey?: string): string { + const base = authToken + ? `${agentUrl}::${createHash('sha256').update(authToken).digest('hex').slice(0, 16)}` + : agentUrl; + return signingCacheKey ? `${base}::${signingCacheKey}` : base; } /** Get a cached connection, refreshing its LRU position. */ @@ -121,7 +123,8 @@ async function getOrCreateConnection( baseUrl: URL, authHeaders: Record, debugLogs: DebugLogEntry[], - label: string + label: string, + signingContext?: AgentSigningContext ): Promise { const cached = getCachedConnection(cacheKey); if (cached) return cached; @@ -129,7 +132,7 @@ async function getOrCreateConnection( const pending = pendingConnections.get(cacheKey); if (pending) return pending; - const promise = connectMCPWithFallback(baseUrl, authHeaders, debugLogs, label) + const promise = connectMCPWithFallback(baseUrl, authHeaders, debugLogs, label, signingContext) .then(client => { connectionCache.set(cacheKey, client); evictLeastRecentlyUsed(); @@ -157,12 +160,13 @@ export async function withCachedConnection( authHeaders: Record, debugLogs: DebugLogEntry[], label: string, - fn: (client: MCPClient) => Promise + fn: (client: MCPClient) => Promise, + signingContext?: AgentSigningContext ): Promise { - const cacheKey = connectionCacheKey(agentUrl, authToken); + const cacheKey = connectionCacheKey(agentUrl, authToken, signingContext?.cacheKey); const baseUrl = new URL(agentUrl); - const mcpClient = await getOrCreateConnection(cacheKey, baseUrl, authHeaders, debugLogs, label); + const mcpClient = await getOrCreateConnection(cacheKey, baseUrl, authHeaders, debugLogs, label, signingContext); try { return await fn(mcpClient); @@ -199,7 +203,14 @@ export async function withCachedConnection( /* ignore */ } - const retryClient = await getOrCreateConnection(cacheKey, baseUrl, authHeaders, debugLogs, `${label} (retry)`); + const retryClient = await getOrCreateConnection( + cacheKey, + baseUrl, + authHeaders, + debugLogs, + `${label} (retry)`, + signingContext + ); try { return await fn(retryClient); @@ -231,6 +242,8 @@ export interface MCPCallOptions { debugLogs?: DebugLogEntry[]; /** Additional headers to send with every request (auth headers take precedence) */ customHeaders?: Record; + /** RFC 9421 signing context — when set, the transport signs outbound ops per seller capability. */ + signingContext?: AgentSigningContext; } /** @@ -259,7 +272,8 @@ export async function connectMCPWithFallback( url: URL, authHeaders: Record, debugLogs: DebugLogEntry[] = [], - label = 'connection' + label = 'connection', + signingContext?: AgentSigningContext ): Promise { return withSpan( 'adcp.mcp.connect', @@ -268,7 +282,7 @@ export async function connectMCPWithFallback( 'adcp.connection_label': label, }, async () => { - return connectMCPWithFallbackImpl(url, authHeaders, debugLogs, label); + return connectMCPWithFallbackImpl(url, authHeaders, debugLogs, label, signingContext); } ); } @@ -277,9 +291,20 @@ async function connectMCPWithFallbackImpl( url: URL, authHeaders: Record, debugLogs: DebugLogEntry[] = [], - label = 'connection' + label = 'connection', + signingContext?: AgentSigningContext ): Promise { - const transportOptions = { requestInit: { headers: authHeaders } }; + const signingFetch = signingContext + ? buildAgentSigningFetch({ + upstream: (input, init) => fetch(input as any, init), + signing: signingContext.signing, + getCapability: signingContext.getCapability, + }) + : undefined; + const transportOptions: StreamableHTTPClientTransportOptions = { + requestInit: { headers: authHeaders }, + ...(signingFetch ? { fetch: signingFetch as typeof fetch } : {}), + }; let failedClient: MCPClient | undefined; try { @@ -372,7 +397,12 @@ async function connectMCPWithFallbackImpl( timestamp: new Date().toISOString(), }); const client = new MCPClient({ name: 'AdCP-Client', version: '1.0.0' }); - await client.connect(new SSEClientTransport(url, { requestInit: { headers: authHeaders } })); + await client.connect( + new SSEClientTransport(url, { + requestInit: { headers: authHeaders }, + ...(signingFetch ? { fetch: signingFetch as typeof fetch } : {}), + }) + ); debugLogs.push({ type: 'success', message: `MCP: Connected via SSE transport for ${label}`, @@ -388,7 +418,8 @@ export async function callMCPTool( args: Record, authToken?: string, debugLogs: DebugLogEntry[] = [], - customHeaders?: Record + customHeaders?: Record, + signingContext?: AgentSigningContext ): Promise { return withSpan( 'adcp.mcp.call_tool', @@ -397,7 +428,7 @@ export async function callMCPTool( 'http.url': agentUrl, }, async () => { - return callMCPToolImpl(agentUrl, toolName, args, authToken, debugLogs, customHeaders); + return callMCPToolImpl(agentUrl, toolName, args, authToken, debugLogs, customHeaders, signingContext); } ); } @@ -412,9 +443,10 @@ export async function callMCPToolRaw( args: Record, authToken?: string, debugLogs: DebugLogEntry[] = [], - customHeaders?: Record + customHeaders?: Record, + signingContext?: AgentSigningContext ): Promise { - return callMCPToolRawImpl(agentUrl, toolName, args, authToken, debugLogs, customHeaders); + return callMCPToolRawImpl(agentUrl, toolName, args, authToken, debugLogs, customHeaders, signingContext); } async function callMCPToolImpl( @@ -423,7 +455,8 @@ async function callMCPToolImpl( args: Record, authToken?: string, debugLogs: DebugLogEntry[] = [], - customHeaders?: Record + customHeaders?: Record, + signingContext?: AgentSigningContext ): Promise { // Inject trace context headers for distributed tracing const traceHeaders = injectTraceHeaders(); @@ -465,7 +498,8 @@ async function callMCPToolImpl( authHeaders, debugLogs, toolName, - client => client.callTool({ name: toolName, arguments: args }) as Promise + client => client.callTool({ name: toolName, arguments: args }) as Promise, + signingContext ); debugLogs.push({ @@ -487,7 +521,8 @@ async function callMCPToolRawImpl( args: Record, authToken?: string, debugLogs: DebugLogEntry[] = [], - customHeaders?: Record + customHeaders?: Record, + signingContext?: AgentSigningContext ): Promise { const traceHeaders = injectTraceHeaders(); const authHeaders = { @@ -496,8 +531,14 @@ async function callMCPToolRawImpl( ...(authToken ? createMCPAuthHeaders(authToken) : {}), }; - return withCachedConnection(agentUrl, authToken, authHeaders, debugLogs, toolName, client => - client.callTool({ name: toolName, arguments: args }) + return withCachedConnection( + agentUrl, + authToken, + authHeaders, + debugLogs, + toolName, + client => client.callTool({ name: toolName, arguments: args }), + signingContext ); } @@ -541,8 +582,9 @@ export async function connectMCP(options: { authProvider?: OAuthClientProvider; debugLogs?: DebugLogEntry[]; customHeaders?: Record; + signingContext?: AgentSigningContext; }): Promise { - const { agentUrl, authToken, authProvider, debugLogs = [], customHeaders } = options; + const { agentUrl, authToken, authProvider, debugLogs = [], customHeaders, signingContext } = options; const baseUrl = new URL(agentUrl); debugLogs.push({ @@ -579,6 +621,17 @@ export async function connectMCP(options: { }); } + // RFC 9421 signing — wrap the transport's fetch so the signer sees the final + // headers the SDK assembled (including any OAuth-issued Authorization) and + // decides per outbound request whether to sign. + if (signingContext) { + transportOptions.fetch = buildAgentSigningFetch({ + upstream: (input, init) => fetch(input as string | URL, init), + signing: signingContext.signing, + getCapability: signingContext.getCapability, + }) as typeof fetch; + } + const transport = new StreamableHTTPClientTransport(baseUrl, transportOptions); try { @@ -618,11 +671,11 @@ export async function connectMCP(options: { * @throws UnauthorizedError if OAuth is required (with transport attached) */ export async function callMCPToolWithOAuth(options: MCPCallOptions): Promise { - const { agentUrl, toolName, args, authToken, authProvider, debugLogs = [], customHeaders } = options; + const { agentUrl, toolName, args, authToken, authProvider, debugLogs = [], customHeaders, signingContext } = options; // If no OAuth provider, use the legacy function if (!authProvider) { - return callMCPTool(agentUrl, toolName, args, authToken, debugLogs, customHeaders); + return callMCPTool(agentUrl, toolName, args, authToken, debugLogs, customHeaders, signingContext); } let client: MCPClient | undefined; @@ -634,6 +687,7 @@ export async function callMCPToolWithOAuth(options: MCPCallOptions): Promise CachedCapability | undefined; + /** Capability cache backing this context (for external invalidation). */ + cache: CapabilityCache; + /** Stable cache key used against the capability cache itself. */ + capabilityCacheKey: string; + /** + * Evict this context's capability entry so the next outbound call + * re-primes `get_adcp_capabilities`. Use after a seller-side rotation + * signal — or on a 401 / protocol-signature error — without having to + * rebuild the cache key from the agent's identifying fields. + */ + invalidate(): void; +} + +/** + * Build an `AgentSigningContext` from an `AgentConfig` when signing is + * configured. Returns `undefined` when the agent has no `request_signing` + * block — callers use this to branch into the no-op fast path. + */ +export function buildAgentSigningContext( + agent: AgentConfig, + options: { cache?: CapabilityCache } = {} +): AgentSigningContext | undefined { + const signing = agent.request_signing; + if (!signing) return undefined; + + const cache = options.cache ?? defaultCapabilityCache; + const keyFingerprint = privateKeyFingerprint(signing); + const capabilityCacheKey = buildCapabilityCacheKey(agent.agent_uri, agent.auth_token, keyFingerprint); + // Transport-connection cache-key suffix binds to a hash of the private key, + // not just the advertised `kid`. Two tenants that misconfigure the same + // `kid` string but hold distinct private keys must not collide on a shared + // cached transport — that would sign one tenant's outbound requests with + // the other tenant's key (same `kid`, different `d`), an impersonation. + const cacheKey = `sig=${keyFingerprint}`; + + return { + signing, + cacheKey, + cache, + capabilityCacheKey, + getCapability: () => cache.get(capabilityCacheKey), + invalidate: () => cache.invalidate(capabilityCacheKey), + }; +} + +/** + * Derive a stable per-key cache-key fragment. Hashes both `kid` and the + * private scalar `d` so that two tenants advertising the same `kid` but + * holding distinct private keys get different cache entries. Truncated to + * 16 hex chars — a collision-resistance budget of 64 bits against random + * keys is plenty for a cache disambiguator, and we never rely on this + * value as a security boundary. + */ +function privateKeyFingerprint(signing: AgentRequestSigningConfig): string { + return createHash('sha256').update(signing.kid).update('\0').update(signing.private_key.d).digest('hex').slice(0, 16); +} diff --git a/src/lib/signing/agent-fetch.ts b/src/lib/signing/agent-fetch.ts new file mode 100644 index 000000000..06537e8d0 --- /dev/null +++ b/src/lib/signing/agent-fetch.ts @@ -0,0 +1,172 @@ +import type { AgentRequestSigningConfig } from '../types/adcp'; +import { createSigningFetch, type CoverContentDigestPredicate } from './fetch'; +import type { CachedCapability, CapabilityCache } from './capability-cache'; +import type { ContentDigestPolicy, VerifierCapability } from './types'; +import type { SignerKey } from './signer'; + +type FetchLike = (input: string | URL | Request, init?: RequestInit) => Promise; + +function bodyToUtf8(body: unknown): string | undefined { + if (body === undefined || body === null) return undefined; + if (typeof body === 'string') return body.length ? body : undefined; + if (body instanceof Uint8Array) return Buffer.from(body).toString('utf8'); + if (body instanceof ArrayBuffer) return Buffer.from(body).toString('utf8'); + // FormData / Blob / ReadableStream / async iterables fall through. Throw + // rather than return `undefined` — a silent pass-through would ship the + // request unsigned with no hint to the caller, defeating the seller's + // `required_for` contract. Matches the strict posture of `createSigningFetch` + // which also refuses unsupported body shapes. + throw new TypeError( + `buildAgentSigningFetch cannot extract an AdCP operation name from a body of type ${describeBody(body)}. The signer only supports string, Uint8Array, and ArrayBuffer bodies because the signature must cover the exact wire bytes.` + ); +} + +function describeBody(body: unknown): string { + if (body && typeof body === 'object') { + const ctor = (body as { constructor?: { name?: string } }).constructor?.name; + if (ctor) return ctor; + } + return typeof body; +} + +/** + * Extract the AdCP operation name from a JSON-RPC request body, if any. + * + * - MCP tool calls: `method === "tools/call"` → `params.name` is the op name. + * - A2A `message/send` / `message/stream`: the op name lives on the first + * data-kind part as `data.skill`. + * - All other JSON-RPC methods (`initialize`, `tools/list`, notifications) + * return `undefined` — those are protocol-layer housekeeping, not AdCP + * operations subject to request-signing policy. + * + * Throws if the body is of a shape the signer can't read (Blob, FormData, + * ReadableStream). The MCP / A2A SDKs both emit JSON strings today; a future + * SDK version switching to streams would silently break signing otherwise. + */ +export function extractAdcpOperation(body: unknown): string | undefined { + const text = bodyToUtf8(body); + if (!text) return undefined; + let parsed: unknown; + try { + parsed = JSON.parse(text); + } catch { + return undefined; + } + if (!parsed || typeof parsed !== 'object') return undefined; + const rpc = parsed as { method?: unknown; params?: unknown }; + + if (rpc.method === 'tools/call') { + const params = rpc.params as { name?: unknown } | undefined; + return typeof params?.name === 'string' ? params.name : undefined; + } + + if (rpc.method === 'message/send' || rpc.method === 'message/stream') { + const params = rpc.params as { message?: { parts?: unknown } } | undefined; + const parts = params?.message?.parts; + if (!Array.isArray(parts)) return undefined; + for (const part of parts) { + if (part && typeof part === 'object') { + const p = part as { kind?: unknown; data?: { skill?: unknown } }; + if (p.kind === 'data' && typeof p.data?.skill === 'string') { + return p.data.skill; + } + } + } + } + + return undefined; +} + +/** + * Decide whether an outbound AdCP call should be signed given the seller's + * advertised capability block and the buyer's override list. + * + * Precedence matches the AdCP spec (`required_for` > `warn_for` > + * `supported_for`): + * 1. `always_sign` on the buyer config — pilot-time override, signs even + * if the seller hasn't listed the op. + * 2. Seller `required_for` — seller rejects unsigned requests, MUST sign. + * 3. Seller `warn_for` — shadow mode. Seller verifies when present and + * logs failures without rejecting; counterparties SHOULD sign so the + * seller can surface failure rates before flipping to `required_for`. + * 4. Seller `supported_for` — sign only if the buyer opted in via + * `sign_supported: true` (defaults off). + * + * Returns false when the capability is unknown (cold cache) except for ops + * in `always_sign`, so the priming `get_adcp_capabilities` call itself is + * never signed. + */ +export function shouldSignOperation( + operation: string | undefined, + capability: VerifierCapability | undefined, + config: AgentRequestSigningConfig +): boolean { + if (!operation) return false; + if (config.always_sign?.includes(operation)) return true; + if (!capability?.supported) return false; + if (capability.required_for?.includes(operation)) return true; + if (capability.warn_for?.includes(operation)) return true; + if (config.sign_supported && capability.supported_for?.includes(operation)) return true; + return false; +} + +/** + * Resolve the seller's content-digest policy into a concrete per-request + * coverage decision. + * + * - `required` → must cover content-digest. + * - `forbidden` → must NOT cover content-digest. + * - `either` / absent → default to covering (body-binding is the safer + * choice; the seller has explicitly allowed both forms). + */ +export function resolveCoverContentDigest(policy: ContentDigestPolicy | undefined): boolean { + if (policy === 'forbidden') return false; + return true; +} + +/** + * Convert an `AgentRequestSigningConfig` into the `SignerKey` shape expected + * by `signRequest` / `createSigningFetch`. + */ +export function toSignerKey(config: AgentRequestSigningConfig): SignerKey { + return { + keyid: config.kid, + alg: config.alg, + privateKey: config.private_key as SignerKey['privateKey'], + }; +} + +export interface BuildAgentSigningFetchOptions { + upstream: FetchLike; + signing: AgentRequestSigningConfig; + /** Lazy accessor for the current cached capability — re-read on every call. */ + getCapability: () => CachedCapability | undefined; +} + +/** + * Build a fetch wrapper suitable for injection into MCP/A2A transports. On + * every outbound request: + * 1. Extract the AdCP operation name from the JSON-RPC body (MCP tool-call + * or A2A message/send). Non-AdCP JSON-RPC methods (e.g., `initialize`) + * pass through unsigned. + * 2. Consult the cached seller capability to decide whether to sign. + * 3. Resolve the seller's content-digest policy into a per-request toggle. + * 4. Delegate to `createSigningFetch` with the decision baked in. + */ +export function buildAgentSigningFetch(options: BuildAgentSigningFetchOptions): FetchLike { + const { upstream, signing, getCapability } = options; + const key = toSignerKey(signing); + + const shouldSign = (_url: string, init: RequestInit | undefined): boolean => { + const operation = extractAdcpOperation(init?.body); + const entry = getCapability(); + return shouldSignOperation(operation, entry?.requestSigning, signing); + }; + + const coverContentDigest: CoverContentDigestPredicate = (_url, _init) => { + const entry = getCapability(); + return resolveCoverContentDigest(entry?.requestSigning?.covers_content_digest); + }; + + return createSigningFetch(upstream, key, { shouldSign, coverContentDigest }); +} diff --git a/src/lib/signing/capability-cache.ts b/src/lib/signing/capability-cache.ts new file mode 100644 index 000000000..2c382b31b --- /dev/null +++ b/src/lib/signing/capability-cache.ts @@ -0,0 +1,119 @@ +import { createHash } from 'node:crypto'; +import type { VerifierCapability } from './types'; + +export interface CachedCapability { + /** RFC 9421 request-signing capability block as advertised by the agent. */ + requestSigning: VerifierCapability | undefined; + /** AdCP major version associated with the capability response, when known. */ + adcpVersion: number | undefined; + /** Epoch seconds when this entry was written. */ + fetchedAt: number; + /** + * Optional explicit epoch-seconds deadline at which this entry becomes + * stale. Overrides the cache's default TTL — used to give negative + * (failed-discovery) entries a shorter refresh window than positive + * entries so a transient seller outage doesn't block signing decisions + * for the full 5-minute TTL. + */ + staleAt?: number; +} + +export interface CapabilityCacheOptions { + /** Seconds before a cached capability is considered stale. Default 300. */ + ttlSeconds?: number; + now?: () => number; +} + +const DEFAULT_TTL_SECONDS = 300; + +/** + * Per-agent cache of the `request_signing` capability block returned by + * `get_adcp_capabilities`. Keyed by a caller-supplied `cacheKey` (typically + * `agent_uri + auth-token-hash`) so that different credentials or URLs get + * independent entries. + * + * Staleness is TTL-based; callers may also invalidate explicitly — e.g. after + * a seller rotates its advertisement mid-session — so the next outbound call + * re-fetches before deciding whether to sign. + */ +export class CapabilityCache { + private readonly entries = new Map(); + /** + * In-flight priming fetches keyed by `cacheKey`. Lives on the instance so + * two `CapabilityCache` objects — e.g., one per tenant in a multi-tenant + * embedding — don't share an `ensureCapabilityLoaded` promise map across + * instances and race each other's writes. The default process-global + * cache uses its own map via `defaultCapabilityCache`. + */ + private readonly inFlight = new Map>(); + private readonly ttlSeconds: number; + private readonly now: () => number; + + constructor(options: CapabilityCacheOptions = {}) { + this.ttlSeconds = options.ttlSeconds ?? DEFAULT_TTL_SECONDS; + this.now = options.now ?? (() => Math.floor(Date.now() / 1000)); + } + + get(cacheKey: string): CachedCapability | undefined { + return this.entries.get(cacheKey); + } + + set(cacheKey: string, entry: CachedCapability): void { + this.entries.set(cacheKey, entry); + } + + invalidate(cacheKey: string): void { + this.entries.delete(cacheKey); + } + + clear(): void { + this.entries.clear(); + this.inFlight.clear(); + } + + isStale(entry: CachedCapability | undefined): boolean { + if (!entry) return true; + const now = this.now(); + if (entry.staleAt !== undefined) return now >= entry.staleAt; + return now - entry.fetchedAt > this.ttlSeconds; + } + + /** @internal Pending-fetch table used by `ensureCapabilityLoaded`. */ + _getInFlight(cacheKey: string): Promise | undefined { + return this.inFlight.get(cacheKey); + } + + /** @internal */ + _setInFlight(cacheKey: string, promise: Promise): void { + this.inFlight.set(cacheKey, promise); + } + + /** @internal */ + _deleteInFlight(cacheKey: string): void { + this.inFlight.delete(cacheKey); + } +} + +/** + * Process-global capability cache. Shared by the ProtocolClient priming path + * and the transport-level signing fetch wrappers so that a single + * `get_adcp_capabilities` call serves every subsequent signing decision for + * an agent. + */ +export const defaultCapabilityCache = new CapabilityCache(); + +/** + * Build a stable cache key from an agent URI, an optional auth-token hash, + * and an optional signer-key fingerprint. Two callers pointing at the same + * agent URI under different signing identities get separate entries — a + * seller can advertise different policies per counterparty key. + * + * Hash is a cache-key disambiguator, not a security boundary; a hypothetical + * collision across users would still transmit only the original caller's + * token (the cache key is not the auth credential itself). + */ +export function buildCapabilityCacheKey(agentUri: string, authToken?: string, signerFingerprint?: string): string { + const tokenSuffix = authToken ? `::${createHash('sha256').update(authToken).digest('hex').slice(0, 16)}` : ''; + const signerSuffix = signerFingerprint ? `::sig=${signerFingerprint}` : ''; + return `${agentUri}${tokenSuffix}${signerSuffix}`; +} diff --git a/src/lib/signing/capability-priming.ts b/src/lib/signing/capability-priming.ts new file mode 100644 index 000000000..028604de1 --- /dev/null +++ b/src/lib/signing/capability-priming.ts @@ -0,0 +1,159 @@ +import type { AgentConfig } from '../types/adcp'; +import type { AgentSigningContext } from './agent-context'; +import type { CachedCapability } from './capability-cache'; +import type { VerifierCapability } from './types'; + +/** + * Op name used to fetch the seller's capability advertisement. The signing + * wrapper short-circuits on this op so the priming request itself is never + * gated by signing. + */ +export const CAPABILITY_OP = 'get_adcp_capabilities'; + +type FetchRaw = (args: Record) => Promise; + +/** + * Extract the `request_signing` capability block from a `get_adcp_capabilities` + * response regardless of how the transport wrapped it: + * + * - MCP `CallToolResult` — `structuredContent` or `content[].text` (JSON). + * - A2A JSON-RPC `SendMessageResponse` — `result` is a `Task` (with + * `artifacts[].parts[].data`) or a `Message` (with `parts[].data`). + * - Already-unwrapped payload — returned as-is. + */ +function extractCapability(response: unknown): { + requestSigning: VerifierCapability | undefined; + adcpVersion: number | undefined; +} { + const payload = unwrapResponse(response); + if (!payload || typeof payload !== 'object') return { requestSigning: undefined, adcpVersion: undefined }; + + const body = payload as Record; + const requestSigning = body.request_signing as VerifierCapability | undefined; + const adcp = body.adcp as { major_versions?: unknown } | undefined; + const versions = Array.isArray(adcp?.major_versions) ? (adcp!.major_versions as unknown[]) : undefined; + const adcpVersion = + versions && versions.length > 0 && typeof versions[0] === 'number' ? (versions[0] as number) : undefined; + + return { requestSigning, adcpVersion }; +} + +function unwrapResponse(response: unknown): unknown { + if (!response || typeof response !== 'object') return response; + const r = response as Record; + + // MCP CallToolResult — structuredContent wins when present. + if (r.structuredContent && typeof r.structuredContent === 'object') return r.structuredContent; + + // MCP CallToolResult — content[].text parsed as JSON. + const content = r.content; + if (Array.isArray(content)) { + for (const chunk of content) { + if (chunk && typeof chunk === 'object' && typeof (chunk as any).text === 'string') { + try { + return JSON.parse((chunk as any).text); + } catch { + // non-JSON text chunk — keep looking + } + } + } + } + + // A2A JSON-RPC response — .result is the Task or Message. + const result = r.result; + if (result && typeof result === 'object') { + // A2A Task: artifacts[0].parts[0].data carries the AdCP payload. + const artifacts = (result as Record).artifacts; + if (Array.isArray(artifacts)) { + for (const artifact of artifacts) { + const parts = (artifact as { parts?: unknown }).parts; + const data = findFirstDataPart(parts); + if (data) return data; + } + } + // A2A Message: parts[0].data directly on the result. + const parts = (result as { parts?: unknown }).parts; + const data = findFirstDataPart(parts); + if (data) return data; + } + + return response; +} + +function findFirstDataPart(parts: unknown): unknown { + if (!Array.isArray(parts)) return undefined; + for (const part of parts) { + if (part && typeof part === 'object') { + const p = part as { kind?: unknown; data?: unknown }; + if (p.kind === 'data' && p.data && typeof p.data === 'object') return p.data; + } + } + return undefined; +} + +/** + * Refresh window applied to a negative-cache entry written after a failed + * discovery call. 60s is short enough that a transient seller outage + * self-heals on the next user action, long enough to avoid pile-ups if the + * seller stays down. + */ +const NEGATIVE_CACHE_TTL_SECONDS = 60; + +/** + * Populate the capability cache for an agent when the `request_signing` entry + * is absent or stale. The injected `fetchRaw` callback is expected to make an + * unsigned `get_adcp_capabilities` call against the counterparty — callers + * wire it to `ProtocolClient.callTool` or the underlying transport helper so + * that no new connection code lives here. + * + * Fails open: if discovery itself fails, we cache an empty entry with a short + * `staleAt` window and return it rather than propagating the error. Signing + * decisions then fall through: + * - Ops in the buyer's `always_sign` list are still signed (with default + * content-digest coverage), so explicit pilot opt-ins keep working. + * - Ops the seller might have listed in `required_for` go out unsigned and + * are rejected with `request_signature_required` at the wire — the user + * sees a clear error rather than an opaque priming wedge, and the next + * retry re-primes. + */ +export async function ensureCapabilityLoaded( + _agent: AgentConfig, + signingContext: AgentSigningContext, + fetchRaw: FetchRaw +): Promise { + const { cache, capabilityCacheKey: key } = signingContext; + const existing = cache.get(key); + if (existing && !cache.isStale(existing)) return existing; + + const pending = cache._getInFlight(key); + if (pending) return pending; + + const promise = fetchRaw({}) + .then(raw => { + const { requestSigning, adcpVersion } = extractCapability(raw); + const entry: CachedCapability = { + requestSigning, + adcpVersion, + fetchedAt: Math.floor(Date.now() / 1000), + }; + cache.set(key, entry); + return entry; + }) + .catch(() => { + const now = Math.floor(Date.now() / 1000); + const entry: CachedCapability = { + requestSigning: undefined, + adcpVersion: undefined, + fetchedAt: now, + staleAt: now + NEGATIVE_CACHE_TTL_SECONDS, + }; + cache.set(key, entry); + return entry; + }) + .finally(() => { + cache._deleteInFlight(key); + }); + + cache._setInFlight(key, promise); + return promise; +} diff --git a/src/lib/signing/client.ts b/src/lib/signing/client.ts new file mode 100644 index 000000000..3824f55d9 --- /dev/null +++ b/src/lib/signing/client.ts @@ -0,0 +1,48 @@ +/** + * Client-side signing surface: what a buyer needs to sign outbound AdCP + * requests per RFC 9421 — signer, canonicalization helpers, fetch wrapper, + * and the capability cache that gates auto-wiring. + * + * Paired with `@adcp/client/signing/server` (verifier / middleware / stores). + * The aggregate `@adcp/client/signing` barrel re-exports both for back-compat. + */ +export { + buildSignatureBase, + canonicalAuthority, + canonicalMethod, + canonicalTargetUri, + formatSignatureParams, + getHeaderValue, + type RequestLike, + type SignatureParams, +} from './canonicalize'; +export { computeContentDigest, contentDigestMatches, parseContentDigest } from './content-digest'; +export { signRequest, type SignedRequest, type SignerKey, type SignRequestOptions } from './signer'; +export { createSigningFetch, type CoverContentDigestPredicate, type SigningFetchOptions } from './fetch'; +export { + ALLOWED_ALGS, + CLOCK_SKEW_TOLERANCE_SECONDS, + MANDATORY_COMPONENTS, + MAX_SIGNATURE_WINDOW_SECONDS, + REQUEST_SIGNING_TAG, + type AdcpJsonWebKey, + type ContentDigestPolicy, + type VerifierCapability, +} from './types'; +export { + CapabilityCache, + buildCapabilityCacheKey, + defaultCapabilityCache, + type CachedCapability, + type CapabilityCacheOptions, +} from './capability-cache'; +export { + buildAgentSigningFetch, + extractAdcpOperation, + resolveCoverContentDigest, + shouldSignOperation, + toSignerKey, + type BuildAgentSigningFetchOptions, +} from './agent-fetch'; +export { buildAgentSigningContext, type AgentSigningContext } from './agent-context'; +export { ensureCapabilityLoaded, CAPABILITY_OP } from './capability-priming'; diff --git a/src/lib/signing/fetch.ts b/src/lib/signing/fetch.ts index d23476f38..d17b52279 100644 --- a/src/lib/signing/fetch.ts +++ b/src/lib/signing/fetch.ts @@ -1,11 +1,29 @@ import { signRequest, type SignerKey, type SignRequestOptions } from './signer'; -export interface SigningFetchOptions extends SignRequestOptions { +/** Callback form for `coverContentDigest` — lets the wrapper decide per call. */ +export type CoverContentDigestPredicate = (url: string, init: RequestInit | undefined) => boolean; + +export interface SigningFetchOptions extends Omit { shouldSign?: (url: string, init: RequestInit | undefined) => boolean; + /** + * Whether to cover `content-digest`. May be a boolean (static) or a + * predicate resolved at signing time against the current request — used by + * the AdCP agent wrapper to honor the seller's `covers_content_digest` + * policy (`required` / `forbidden` / `either`) per operation. + */ + coverContentDigest?: boolean | CoverContentDigestPredicate; } type FetchLike = (input: string | URL | Request, init?: RequestInit) => Promise; +/** + * Header names whose wire values are produced by the signer itself. Any + * caller-supplied value gets stripped before signing so a misconfigured + * custom-headers bag can't silently overwrite (or bypass) the RFC 9421 + * signature output. + */ +const SIGNING_RESERVED_HEADERS = new Set(['signature', 'signature-input', 'content-digest']); + export function createSigningFetch(upstream: FetchLike, key: SignerKey, options: SigningFetchOptions = {}): FetchLike { return async (input, init) => { const url = typeof input === 'string' ? input : input instanceof URL ? input.toString() : input.url; @@ -20,13 +38,25 @@ export function createSigningFetch(upstream: FetchLike, key: SignerKey, options: } const headers = headersToRecord(init?.headers); + for (const name of Object.keys(headers)) { + if (SIGNING_RESERVED_HEADERS.has(name.toLowerCase())) { + delete headers[name]; + } + } const hasContentType = Object.keys(headers).some(k => k.toLowerCase() === 'content-type'); if (!hasContentType && (init?.body !== undefined || method !== 'GET')) { headers['content-type'] = 'application/json'; } const body = bodyToString(init?.body); - const signed = signRequest({ method, url, headers, body }, key, options); + const coverContentDigest = + typeof options.coverContentDigest === 'function' + ? options.coverContentDigest(url, init) + : options.coverContentDigest; + const { coverContentDigest: _omit, ...signerOptionsBase } = options; + const signerOptions: SignRequestOptions = { ...signerOptionsBase, coverContentDigest }; + + const signed = signRequest({ method, url, headers, body }, key, signerOptions); const mergedInit: RequestInit = { ...init, method, headers: signed.headers }; if (body !== undefined && mergedInit.body === undefined) mergedInit.body = body; diff --git a/src/lib/signing/index.ts b/src/lib/signing/index.ts index 510ea2fd8..49a1a1f0c 100644 --- a/src/lib/signing/index.ts +++ b/src/lib/signing/index.ts @@ -1,38 +1,15 @@ -export { - buildSignatureBase, - canonicalAuthority, - canonicalMethod, - canonicalTargetUri, - formatSignatureParams, - getHeaderValue, - type RequestLike, - type SignatureParams, -} from './canonicalize'; -export { computeContentDigest, contentDigestMatches, parseContentDigest } from './content-digest'; -export { jwkToPublicKey, verifySignature } from './crypto'; -export { RequestSignatureError, type RequestSignatureErrorCode } from './errors'; -export { StaticJwksResolver, type JwksResolver } from './jwks'; -export { parseSignature, parseSignatureInput, type ParsedSignature, type ParsedSignatureInput } from './parser'; -export { - InMemoryReplayStore, - type InMemoryReplayStoreOptions, - type ReplayInsertResult, - type ReplayStore, -} from './replay'; -export { InMemoryRevocationStore, type RevocationStore } from './revocation'; -export { - ALLOWED_ALGS, - CLOCK_SKEW_TOLERANCE_SECONDS, - MANDATORY_COMPONENTS, - MAX_SIGNATURE_WINDOW_SECONDS, - REQUEST_SIGNING_TAG, - type AdcpJsonWebKey, - type ContentDigestPolicy, - type RevocationSnapshot, - type VerifiedSigner, - type VerifierCapability, -} from './types'; -export { verifyRequestSignature, type VerifyRequestOptions } from './verifier'; -export { signRequest, type SignedRequest, type SignerKey, type SignRequestOptions } from './signer'; -export { createSigningFetch, type SigningFetchOptions } from './fetch'; -export { createExpressVerifier, type ExpressLike, type ExpressMiddlewareOptions } from './middleware'; +/** + * Aggregate signing barrel. Re-exports the `client` and `server` sub-barrels + * so existing consumers of `@adcp/client/signing` keep working. + * + * New code should import from the narrower surface that matches its role: + * + * import { createSigningFetch } from '@adcp/client/signing/client'; + * import { createExpressVerifier } from '@adcp/client/signing/server'; + * + * — coding agents reading a file cold only need to hold one half of the + * taxonomy, not all 30+ symbols. This aggregate barrel remains the stable + * entry point until a future major release drops it. + */ +export * from './client'; +export * from './server'; diff --git a/src/lib/signing/server.ts b/src/lib/signing/server.ts new file mode 100644 index 000000000..62824552b --- /dev/null +++ b/src/lib/signing/server.ts @@ -0,0 +1,46 @@ +/** + * Server-side signing surface: what a seller running an AdCP agent needs to + * verify inbound RFC 9421 signatures — verifier pipeline, Express-shaped + * middleware, pluggable JWKS / replay / revocation stores, and the error + * taxonomy. + * + * Paired with `@adcp/client/signing/client` (signer / fetch wrapper / + * capability cache). The aggregate `@adcp/client/signing` barrel re-exports + * both for back-compat. + */ +export { + buildSignatureBase, + canonicalAuthority, + canonicalMethod, + canonicalTargetUri, + formatSignatureParams, + getHeaderValue, + type RequestLike, + type SignatureParams, +} from './canonicalize'; +export { computeContentDigest, contentDigestMatches, parseContentDigest } from './content-digest'; +export { jwkToPublicKey, verifySignature } from './crypto'; +export { RequestSignatureError, type RequestSignatureErrorCode } from './errors'; +export { StaticJwksResolver, type JwksResolver } from './jwks'; +export { parseSignature, parseSignatureInput, type ParsedSignature, type ParsedSignatureInput } from './parser'; +export { + InMemoryReplayStore, + type InMemoryReplayStoreOptions, + type ReplayInsertResult, + type ReplayStore, +} from './replay'; +export { InMemoryRevocationStore, type RevocationStore } from './revocation'; +export { + ALLOWED_ALGS, + CLOCK_SKEW_TOLERANCE_SECONDS, + MANDATORY_COMPONENTS, + MAX_SIGNATURE_WINDOW_SECONDS, + REQUEST_SIGNING_TAG, + type AdcpJsonWebKey, + type ContentDigestPolicy, + type RevocationSnapshot, + type VerifiedSigner, + type VerifierCapability, +} from './types'; +export { verifyRequestSignature, type VerifyRequestOptions } from './verifier'; +export { createExpressVerifier, type ExpressLike, type ExpressMiddlewareOptions } from './middleware'; diff --git a/src/lib/signing/types.ts b/src/lib/signing/types.ts index 4fa719850..bc1a73a1c 100644 --- a/src/lib/signing/types.ts +++ b/src/lib/signing/types.ts @@ -4,6 +4,14 @@ export interface VerifierCapability { supported: boolean; covers_content_digest: ContentDigestPolicy; required_for: string[]; + /** + * Shadow-mode bridge between `supported_for` and `required_for`: the seller + * verifies signatures when present and logs failures but does NOT reject + * unsigned requests. Counterparties SHOULD sign ops in this list so sellers + * can surface failure rates before flipping to `required_for`. Precedence: + * `required_for` > `warn_for` > `supported_for`. + */ + warn_for?: string[]; supported_for?: string[]; } diff --git a/src/lib/types/adcp.ts b/src/lib/types/adcp.ts index f651b64a2..2c651e14f 100644 --- a/src/lib/types/adcp.ts +++ b/src/lib/types/adcp.ts @@ -220,6 +220,73 @@ export interface AgentOAuthClient { client_secret_expires_at?: number; } +/** + * Private JWK carrying the `d` scalar required to sign. Narrower than the + * generic JWK shape to give hand-authors a compiler error when they paste + * the public JWK (which lacks `d`) by accident. + */ +export interface AdcpPrivateJsonWebKey { + kid: string; + kty: string; + crv?: string; + alg?: string; + use?: string; + key_ops?: string[]; + x?: string; + y?: string; + /** Private scalar. Required for signing. */ + d: string; + [extra: string]: unknown; +} + +/** + * Request-signing configuration for an agent. When present on an AgentConfig, + * outbound MCP/A2A calls are gated by the seller's advertised + * `request_signing` capability block (fetched once via `get_adcp_capabilities` + * and cached): operations listed in `required_for` / `supported_for` (or + * `always_sign`) are signed with this key per RFC 9421. + * + * Content-digest coverage is resolved per request from the seller's + * advertised `covers_content_digest` policy: `required` covers, `forbidden` + * omits, `either` (or absent) covers by default — body-binding is the safer + * choice and a seller advertising `either` has explicitly allowed both forms. + */ +export interface AgentRequestSigningConfig { + /** Key identifier (published by the buyer at its JWKS endpoint) */ + kid: string; + /** Signature algorithm. Must match the key material. */ + alg: 'ed25519' | 'ecdsa-p256-sha256'; + /** + * Private signing key as a JWK. Must include `d` (the private scalar); + * other fields mirror the public JWK the buyer publishes for verification. + */ + private_key: AdcpPrivateJsonWebKey; + /** + * Agent's base URL (e.g., `https://buyer.example.com`). Consistent with + * every other `agent_url` field across AdCP — formats, creatives, signals, + * brand. Sellers derive the JWKS endpoint via the conventional well-known + * suffix (`{agent_url}/.well-known/adcp-jwks.json`); a handful of deploys + * override the path via seller-side resolver configuration. + * + * The client does not read this field — the signer only needs the local + * private key — but it's carried here so the field sits next to `kid` / + * `alg` / `private_key` and is available to downstream audit logging or + * custom verifier wiring. + */ + agent_url: string; + /** + * AdCP operation names to sign regardless of the seller's advertisement. + * Useful during pilots before a counterparty flips an op into `required_for`. + */ + always_sign?: string[]; + /** + * When true, also sign operations the seller lists in `supported_for` (but + * not `required_for`). Defaults to false — conservative "sign what the + * seller asks for" behavior. + */ + sign_supported?: boolean; +} + // Agent Configuration Types export interface AgentConfig { id: string; @@ -265,6 +332,14 @@ export interface AgentConfig { * ``` */ headers?: Record; + + /** + * Optional — when set, outbound requests to this agent are signed per + * RFC 9421 for operations the agent advertises in its `request_signing` + * capability block (fetched once via `get_adcp_capabilities` and cached + * by the client). See {@link AgentRequestSigningConfig}. + */ + request_signing?: AgentRequestSigningConfig; } // Testing Types diff --git a/test/request-signing-agent-a2a.test.js b/test/request-signing-agent-a2a.test.js new file mode 100644 index 000000000..7ebfe7101 --- /dev/null +++ b/test/request-signing-agent-a2a.test.js @@ -0,0 +1,241 @@ +const { test } = require('node:test'); +const assert = require('node:assert'); +const http = require('node:http'); +const { readFileSync } = require('node:fs'); +const path = require('node:path'); + +const { ProtocolClient } = require('../dist/lib/protocols/index.js'); +const { closeConnections } = require('../dist/lib/protocols/index.js'); +const { defaultCapabilityCache } = require('../dist/lib/signing/client.js'); + +const KEYS_PATH = path.join( + __dirname, + '..', + 'compliance', + 'cache', + 'latest', + 'test-vectors', + 'request-signing', + 'keys.json' +); + +const keys = JSON.parse(readFileSync(KEYS_PATH, 'utf8')).keys; +const ed = keys.find(k => k.kid === 'test-ed25519-2026'); +const privateJwk = { ...ed, d: ed._private_d_for_test_only }; +delete privateJwk._private_d_for_test_only; +delete privateJwk.key_ops; +delete privateJwk.use; + +/** + * Minimal A2A-speaking stub. Implements two endpoints: + * + * - GET `/.well-known/agent.json` — returns an AgentCard whose `url` points + * back at `/rpc` on the same host, so the A2A client sends JSON-RPC + * `message/send` calls there. + * - POST `/rpc` — accepts JSON-RPC; for `message/send` it pulls the first + * data-kind part (which is where AdCP puts `{skill, parameters}`) and + * synthesizes a completed Task result whose artifact echoes JSON payload. + * + * Every inbound JSON-RPC request is recorded alongside the skill the + * handler observed, so tests can assert whether the Signature-Input / + * Signature / Content-Digest headers were present per skill. + */ +async function startA2aStub(initialCapability) { + const state = { + capability: initialCapability, + rpcCalls: [], + }; + + const httpServer = http.createServer(async (req, res) => { + if (req.method === 'GET' && req.url === '/.well-known/agent.json') { + const { port } = httpServer.address(); + const card = { + protocolVersion: '0.3.0', + name: 'signing-a2a-stub', + description: 'A2A stub for request-signing integration tests', + url: `http://127.0.0.1:${port}/rpc`, + preferredTransport: 'JSONRPC', + version: '1.0.0', + defaultInputModes: ['application/json'], + defaultOutputModes: ['application/json'], + capabilities: { streaming: false, pushNotifications: false }, + skills: [ + { + id: 'get_adcp_capabilities', + name: 'get_adcp_capabilities', + description: 'AdCP capability discovery', + tags: ['adcp'], + }, + { id: 'create_media_buy', name: 'create_media_buy', description: 'AdCP op', tags: ['adcp'] }, + { id: 'another_op', name: 'another_op', description: 'AdCP op', tags: ['adcp'] }, + ], + }; + res.statusCode = 200; + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify(card)); + return; + } + + if (req.method !== 'POST' || req.url !== '/rpc') { + res.statusCode = 404; + res.end('not found'); + return; + } + + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const bodyBuf = Buffer.concat(chunks); + let parsed; + try { + parsed = JSON.parse(bodyBuf.toString('utf8')); + } catch { + res.statusCode = 400; + res.end('bad json'); + return; + } + + const skill = + parsed?.params?.message?.parts?.find(p => p?.kind === 'data' && typeof p?.data?.skill === 'string')?.data + ?.skill ?? ''; + + state.rpcCalls.push({ headers: { ...req.headers }, skill, method: parsed.method }); + + const resultPayload = + skill === 'get_adcp_capabilities' + ? { + adcp: { major_versions: [3] }, + supported_protocols: ['media_buy'], + request_signing: state.capability, + } + : { ok: true }; + + const response = { + jsonrpc: '2.0', + id: parsed.id, + result: { + id: `task_${Date.now()}`, + contextId: `ctx_${Date.now()}`, + kind: 'task', + status: { state: 'completed', timestamp: new Date().toISOString() }, + history: [parsed.params.message], + artifacts: [ + { + artifactId: 'artifact-1', + parts: [{ kind: 'data', data: resultPayload }], + }, + ], + }, + }; + + res.statusCode = 200; + res.setHeader('content-type', 'application/json'); + res.end(JSON.stringify(response)); + }); + + await new Promise(resolve => httpServer.listen(0, '127.0.0.1', resolve)); + const addr = httpServer.address(); + return { + url: `http://127.0.0.1:${addr.port}`, + state, + stop: () => { + if (typeof httpServer.closeAllConnections === 'function') { + httpServer.closeAllConnections(); + } + return new Promise(resolve => httpServer.close(() => resolve())); + }, + }; +} + +function agentFor(url) { + return { + id: 'test-agent-a2a', + name: 'Test A2A Agent', + agent_uri: url, + protocol: 'a2a', + request_signing: { + kid: 'test-ed25519-2026', + alg: 'ed25519', + private_key: privateJwk, + agent_url: 'https://buyer.example.com', + }, + }; +} + +async function resetGlobalState() { + await closeConnections('a2a'); + defaultCapabilityCache.clear(); +} + +async function cleanup(stub) { + await closeConnections('a2a'); + await stub.stop(); +} + +test('A2A: priming get_adcp_capabilities is sent unsigned', async () => { + await resetGlobalState(); + const stub = await startA2aStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'get_adcp_capabilities', {}); + const caps = stub.state.rpcCalls.filter(r => r.skill === 'get_adcp_capabilities'); + assert.ok(caps.length >= 1, 'at least one get_adcp_capabilities hit the stub'); + for (const call of caps) { + assert.strictEqual(call.headers['signature-input'], undefined, 'discovery is never signed on A2A'); + } + } finally { + await cleanup(stub); + } +}); + +test('A2A: create_media_buy in required_for gets signed — skill extracted from message/send body', async () => { + await resetGlobalState(); + const stub = await startA2aStub({ + supported: true, + covers_content_digest: 'required', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'create_media_buy', { plan_id: 'plan_a2a_001' }); + const cmb = stub.state.rpcCalls.filter(r => r.skill === 'create_media_buy'); + assert.strictEqual(cmb.length, 1, 'one create_media_buy reached the stub'); + const headers = cmb[0].headers; + assert.match(headers['signature-input'] || '', /^sig1=/, 'Signature-Input present on A2A signed call'); + assert.match(headers['signature'] || '', /^sig1=:/, 'Signature present with sig1 label'); + assert.match( + headers['content-digest'] || '', + /sha-256=/, + 'Content-Digest present (covers_content_digest: required)' + ); + assert.match( + headers['signature-input'], + /"content-digest"/, + 'Signature-Input lists content-digest as covered component' + ); + } finally { + await cleanup(stub); + } +}); + +test('A2A: ops outside the seller advertisement pass through unsigned', async () => { + await resetGlobalState(); + const stub = await startA2aStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'another_op', {}); + const call = stub.state.rpcCalls.filter(r => r.skill === 'another_op')[0]; + assert.ok(call, 'another_op reached the stub'); + assert.strictEqual(call.headers['signature-input'], undefined, 'another_op unsigned on A2A'); + } finally { + await cleanup(stub); + } +}); + +test('A2A: teardown', async () => { + await resetGlobalState(); +}); diff --git a/test/request-signing-agent-integration.test.js b/test/request-signing-agent-integration.test.js new file mode 100644 index 000000000..144cf2ad9 --- /dev/null +++ b/test/request-signing-agent-integration.test.js @@ -0,0 +1,416 @@ +const { test } = require('node:test'); +const assert = require('node:assert'); +const http = require('node:http'); +const { readFileSync } = require('node:fs'); +const path = require('node:path'); + +const { McpServer } = require('@modelcontextprotocol/sdk/server/mcp.js'); +const { StreamableHTTPServerTransport } = require('@modelcontextprotocol/sdk/server/streamableHttp.js'); + +const { ProtocolClient } = require('../dist/lib/protocols/index.js'); +const { closeMCPConnections } = require('../dist/lib/protocols/mcp.js'); +const { defaultCapabilityCache, buildAgentSigningContext } = require('../dist/lib/signing/client.js'); + +const KEYS_PATH = path.join( + __dirname, + '..', + 'compliance', + 'cache', + 'latest', + 'test-vectors', + 'request-signing', + 'keys.json' +); + +const keys = JSON.parse(readFileSync(KEYS_PATH, 'utf8')).keys; +const ed = keys.find(k => k.kid === 'test-ed25519-2026'); +const privateJwk = { ...ed, d: ed._private_d_for_test_only }; +delete privateJwk._private_d_for_test_only; +delete privateJwk.key_ops; +delete privateJwk.use; + +/** + * Minimal MCP-speaking stub. Records the raw headers from each inbound + * tool-call request so tests can assert whether Signature-Input / Signature / + * Content-Digest headers are present. The server's advertised capability is + * mutable so tests can exercise the "seller rotates required_for mid-session" + * case. Every inbound POST is tagged with the tool name the handler observes + * by closing `entry` over the per-request MCP server instance. + */ +async function startMcpStub(initialCapability) { + const state = { + capability: initialCapability, + toolCallHeaders: [], + }; + + const createServer = entry => { + const mcp = new McpServer({ name: 'signing-stub', version: '1.0.0' }); + + mcp.tool('get_adcp_capabilities', {}, async () => { + entry.toolName = 'get_adcp_capabilities'; + return { + content: [ + { + type: 'text', + text: JSON.stringify({ + adcp: { major_versions: [3] }, + supported_protocols: ['media_buy'], + request_signing: state.capability, + }), + }, + ], + }; + }); + + const echoAs = name => async () => { + entry.toolName = name; + return { content: [{ type: 'text', text: JSON.stringify({ ok: true }) }] }; + }; + + mcp.tool('create_media_buy', {}, echoAs('create_media_buy')); + mcp.tool('another_op', {}, echoAs('another_op')); + mcp.tool('unsigned_op', {}, echoAs('unsigned_op')); + + return mcp; + }; + + const httpServer = http.createServer(async (req, res) => { + if (!req.url || (req.url !== '/mcp' && req.url !== '/mcp/')) { + res.statusCode = 404; + res.end('not found'); + return; + } + + const entry = { headers: { ...req.headers }, toolName: undefined }; + state.toolCallHeaders.push(entry); + + const mcp = createServer(entry); + const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: undefined }); + try { + await mcp.connect(transport); + await transport.handleRequest(req, res); + } finally { + await mcp.close(); + } + }); + + await new Promise(resolve => httpServer.listen(0, '127.0.0.1', resolve)); + const addr = httpServer.address(); + return { + url: `http://127.0.0.1:${addr.port}/mcp`, + state, + stop: () => { + // MCP clients hold a persistent connection; close it forcibly so + // httpServer.close() doesn't wait for the keep-alive to drain. + if (typeof httpServer.closeAllConnections === 'function') { + httpServer.closeAllConnections(); + } + return new Promise(resolve => httpServer.close(() => resolve())); + }, + }; +} + +function agentFor(url) { + return { + id: 'test-agent', + name: 'Test Agent', + agent_uri: url, + protocol: 'mcp', + request_signing: { + kid: 'test-ed25519-2026', + alg: 'ed25519', + private_key: privateJwk, + agent_url: 'https://buyer.example.com', + }, + }; +} + +async function resetGlobalState() { + await closeMCPConnections(); + defaultCapabilityCache.clear(); +} + +async function cleanup(stub) { + await closeMCPConnections(); + await stub.stop(); +} + +test('priming: unsigned get_adcp_capabilities succeeds when required_for does not cover it', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + const result = await ProtocolClient.callTool(agentFor(stub.url), 'get_adcp_capabilities', {}); + assert.ok(result, 'capabilities call returned a response'); + + const capsCalls = stub.state.toolCallHeaders.filter(r => r.toolName === 'get_adcp_capabilities'); + assert.ok(capsCalls.length >= 1, 'at least one get_adcp_capabilities tools/call hit the stub'); + for (const call of capsCalls) { + assert.strictEqual( + call.headers['signature-input'], + undefined, + 'get_adcp_capabilities must be sent unsigned — it is the discovery op' + ); + } + } finally { + await cleanup(stub); + } +}); + +test('create_media_buy in required_for gets signed with Signature-Input / Signature / Content-Digest', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'required', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'create_media_buy', { plan_id: 'plan_001' }); + const cmbCalls = stub.state.toolCallHeaders.filter(r => r.toolName === 'create_media_buy'); + assert.strictEqual(cmbCalls.length, 1, 'one create_media_buy tools/call hit the stub'); + const headers = cmbCalls[0].headers; + assert.match(headers['signature-input'] || '', /^sig1=/, 'Signature-Input is present with sig1 label'); + assert.match(headers['signature'] || '', /^sig1=:/, 'Signature header is present with sig1 label'); + assert.ok(headers['content-digest'], 'Content-Digest header is present (covers_content_digest: required)'); + assert.match(headers['content-digest'], /sha-256=/, 'Content-Digest is sha-256'); + } finally { + await cleanup(stub); + } +}); + +test('ops outside required_for / supported_for / always_sign pass through unsigned', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'unsigned_op', {}); + const calls = stub.state.toolCallHeaders.filter(r => r.toolName === 'unsigned_op'); + assert.strictEqual(calls.length, 1); + assert.strictEqual(calls[0].headers['signature-input'], undefined); + } finally { + await cleanup(stub); + } +}); + +test('covers_content_digest: forbidden → signer omits content-digest coverage', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'forbidden', + required_for: ['create_media_buy'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'create_media_buy', { plan_id: 'plan_001' }); + const cmb = stub.state.toolCallHeaders.filter(r => r.toolName === 'create_media_buy')[0]; + assert.ok(cmb.headers['signature-input'], 'request is still signed'); + assert.strictEqual( + cmb.headers['content-digest'], + undefined, + "Content-Digest MUST NOT be attached when seller policy is 'forbidden'" + ); + // And the Signature-Input MUST NOT list content-digest as a covered component. + assert.doesNotMatch( + cmb.headers['signature-input'], + /"content-digest"/, + 'Signature-Input does not cover content-digest' + ); + } finally { + await cleanup(stub); + } +}); + +test('capability rotation: seller adds op to required_for → cache refresh picks it up', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + const agent = agentFor(stub.url); + + // First call: another_op is NOT in required_for → unsigned. + await ProtocolClient.callTool(agent, 'another_op', {}); + let call = stub.state.toolCallHeaders.filter(r => r.toolName === 'another_op')[0]; + assert.strictEqual(call.headers['signature-input'], undefined, 'another_op sent unsigned initially'); + + // Seller rotates capability. + stub.state.capability = { + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy', 'another_op'], + }; + // Simulate the cache TTL expiry / explicit invalidation that would + // force a re-fetch on the next outbound call. Use the context's own + // invalidate() so the test doesn't reach back into the cache-key shape. + buildAgentSigningContext(agent).invalidate(); + + // Second call: capability re-fetched, another_op now in required_for → signed. + await ProtocolClient.callTool(agent, 'another_op', {}); + const newerCall = stub.state.toolCallHeaders.filter(r => r.toolName === 'another_op').slice(-1)[0]; + assert.ok( + newerCall.headers['signature-input'], + 'another_op is signed after capability rotation + cache invalidation' + ); + } finally { + await cleanup(stub); + } +}); + +test('always_sign override forces signing even when seller has not listed the op', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + const agent = agentFor(stub.url); + agent.request_signing.always_sign = ['another_op']; + await ProtocolClient.callTool(agent, 'another_op', {}); + const call = stub.state.toolCallHeaders.filter(r => r.toolName === 'another_op')[0]; + assert.ok(call.headers['signature-input'], 'always_sign forces signing'); + } finally { + await cleanup(stub); + } +}); + +test('warn_for: shadow-mode op is signed so the seller can surface failure rates', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: [], + warn_for: ['another_op'], + }); + try { + await ProtocolClient.callTool(agentFor(stub.url), 'another_op', {}); + const call = stub.state.toolCallHeaders.filter(r => r.toolName === 'another_op')[0]; + assert.ok(call.headers['signature-input'], 'warn_for ops SHOULD be signed so sellers get shadow-mode telemetry'); + } finally { + await cleanup(stub); + } +}); + +test('customHeaders signing-reserved keys are stripped before signing', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + const agent = agentFor(stub.url); + // Malicious or misconfigured caller tries to pre-set Signature-Input + + // Content-Digest. The signer must overwrite both — a silent pass-through + // would break verification at the seller. + agent.headers = { + 'signature-input': 'sig1=("@method");created=0;expires=0;keyid="attacker";alg="ed25519"', + 'content-digest': 'sha-256=:AAAA:', + 'x-benign-header': 'yes', + }; + await ProtocolClient.callTool(agent, 'create_media_buy', {}); + const call = stub.state.toolCallHeaders.filter(r => r.toolName === 'create_media_buy')[0]; + assert.match(call.headers['signature-input'], /keyid="test-ed25519-2026"/, 'signer produced the Signature-Input'); + assert.doesNotMatch( + call.headers['signature-input'], + /keyid="attacker"/, + 'attacker-supplied Signature-Input was overwritten' + ); + assert.match( + call.headers['content-digest'], + /sha-256=:[^A]/, + 'signer recomputed Content-Digest from the real body' + ); + assert.strictEqual(call.headers['x-benign-header'], 'yes', 'non-reserved customHeaders still pass through'); + } finally { + await cleanup(stub); + } +}); + +test('concurrent cold-cache calls share a single get_adcp_capabilities fetch', async () => { + await resetGlobalState(); + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + try { + const agent = agentFor(stub.url); + // Fire three concurrent create_media_buy calls against a cold cache. + // The priming dedupe in capability-priming.ts must collapse them to + // exactly one get_adcp_capabilities fetch — otherwise a seller with + // low quota for discovery could be hammered every time a caller + // fans out. + await Promise.all([ + ProtocolClient.callTool(agent, 'create_media_buy', {}), + ProtocolClient.callTool(agent, 'create_media_buy', {}), + ProtocolClient.callTool(agent, 'create_media_buy', {}), + ]); + const primingCalls = stub.state.toolCallHeaders.filter(r => r.toolName === 'get_adcp_capabilities'); + assert.strictEqual(primingCalls.length, 1, 'priming dedupe folded 3 cold-cache calls into 1 discovery call'); + } finally { + await cleanup(stub); + } +}); + +test('priming failure → fail-open: next call proceeds unsigned, always_sign still forces signing', async () => { + await resetGlobalState(); + // Stand up a stub that rejects get_adcp_capabilities with a 500-equivalent + // error. create_media_buy still succeeds. The client should NOT wedge — + // it should cache a negative entry and let the call proceed. + const stub = await startMcpStub({ + supported: true, + covers_content_digest: 'either', + required_for: ['create_media_buy'], + }); + // Monkey-patch the capability to throw. + Object.defineProperty(stub.state, 'capability', { + get() { + throw new Error('simulated discovery outage'); + }, + }); + try { + const agent = agentFor(stub.url); + agent.request_signing.always_sign = ['create_media_buy']; + // Fail-open: the call must not throw on priming failure, and always_sign + // ops still get signed with sensible content-digest defaults. + await ProtocolClient.callTool(agent, 'create_media_buy', {}); + const call = stub.state.toolCallHeaders.filter(r => r.toolName === 'create_media_buy')[0]; + assert.ok(call.headers['signature-input'], 'always_sign forces signing even when seller discovery failed'); + } finally { + await cleanup(stub); + } +}); + +test('ensureCapabilityLoaded: network-level fetch rejection → 60s negative cache', async () => { + await resetGlobalState(); + const { ensureCapabilityLoaded } = require('../dist/lib/signing/client.js'); + // Synthesize an agent + signing context without a live server so we can + // drive ensureCapabilityLoaded with a rejecting fetchRaw directly — the + // integration test above only exercises the error-response path (tool + // handler throws → CallToolResult.isError). This test covers the + // transport-level rejection that maps to .catch in capability-priming.ts. + const agent = agentFor('http://127.0.0.1:0/mcp'); + const signingContext = buildAgentSigningContext(agent); + const before = Math.floor(Date.now() / 1000); + const entry = await ensureCapabilityLoaded(agent, signingContext, async () => { + throw new Error('ECONNREFUSED'); + }); + const after = Math.floor(Date.now() / 1000); + assert.strictEqual(entry.requestSigning, undefined, 'failed priming caches an empty capability'); + assert.ok(entry.staleAt !== undefined, 'failed priming sets an explicit staleAt'); + const window = entry.staleAt - entry.fetchedAt; + assert.strictEqual(window, 60, 'negative-cache window is 60s (shorter than the 300s positive TTL)'); + assert.ok(entry.fetchedAt >= before && entry.fetchedAt <= after, 'fetchedAt is "now"'); +}); + +test('teardown: close pooled MCP connections', async () => { + await resetGlobalState(); +});