From f634ad04391ac602012ca942ac3e09d87c6b3f55 Mon Sep 17 00:00:00 2001 From: Mark Mennell Date: Thu, 17 Sep 2026 16:52:24 +0800 Subject: [PATCH] Add caller-bound renewable token providers --- AGENTS.md | 3 +- CHANGELOG.md | 4 + README.md | 5 + ROADMAP.md | 5 + docs/token-providers.md | 93 ++++++++++++ package.json | 1 + src/client/client.ts | 184 ++++++++++++------------ src/client/errors.ts | 40 ++++++ src/client/index.ts | 1 + src/client/token-provider.ts | 69 +++++++++ src/client/types.ts | 8 +- src/client/ws.ts | 5 +- src/context.ts | 4 +- src/wait.ts | 4 +- test/fake-fmsg-server.ts | 5 + test/http.test.ts | 8 +- test/token-provider.test.ts | 269 +++++++++++++++++++++++++++++++++++ 17 files changed, 598 insertions(+), 110 deletions(-) create mode 100644 docs/token-providers.md create mode 100644 src/client/errors.ts create mode 100644 src/client/token-provider.ts create mode 100644 test/token-provider.test.ts diff --git a/AGENTS.md b/AGENTS.md index 781ea40..662f903 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -11,7 +11,8 @@ The canonical contract is the fmsg-webapi README and `src/client/` is written against it directly. Key facts: - `POST /fmsg/token` exchanges an `fmsgk_…` API key for a short-lived JWT whose `sub` is the address. - The client refreshes it 5 minutes before expiry and retries once on 401. + The client also accepts a caller-bound `TokenProvider` (see `docs/token-providers.md`). It renews + 5 minutes before expiry, capped at half the acquired lifetime, and retries once on 401. - `id`/`pid` are int64 JSON numbers. They are decimal **strings** everywhere in this codebase; only `src/client/message-id.ts` converts at the JSON boundary (reviver with `context.source`). - Sending is draft → attach → send; the draft is deleted if a later step fails. diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d1540c..8196a5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,10 @@ This is the next planned release; publication still happens through a `v0.2.0` G ### Fixes and improvements +- Accept caller-bound `TokenProvider` implementations in the client library alongside API keys. + Share renewal across concurrent requests, pin the address, bound acquisition time, and propagate + cancellation. Cap early renewal for short-lived tokens and reuse renewal after late 401 responses. + This is the OAuth foundation; hosted OAuth remains separate integration work. - Retry protected reads when a WebSocket announces a message before it is readable. If retries run out, schedule a delayed inbox catch-up without requiring another push. Fix pre-cancelled waits and preserve request deadlines. diff --git a/README.md b/README.md index ff5178c..85d8357 100644 --- a/README.md +++ b/README.md @@ -190,6 +190,11 @@ client.close(); Its result includes the replacement count (`redactions`) and transmitted `topic`. Attachments are unchanged. Use `streamAttachment()` to consume large files incrementally; consume or cancel its stream. +Applications with their own authorization integration can pass a `TokenProvider` instead of an +API-key string. The client shares renewal across concurrent requests and keeps the authenticated +address fixed. See the [token-provider contract](./docs/token-providers.md). This library interface +does not enable hosted OAuth in the MCP executable yet. + ## Development ```sh diff --git a/ROADMAP.md b/ROADMAP.md index da5b9a0..2822498 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -29,6 +29,11 @@ Support claims must name tested clients and versions; agents without MCP need an Ship A first, then B–D. Plan E with the host/account-system maintainer. Release work in F can proceed earlier; hosted-service promises depend on verified OAuth and operational behavior. +E's [client token-provider foundation](docs/token-providers.md) supports renewable upstream credentials +and local isolation tests. Incoming OAuth validation, discovery, token exchange and actual hosted-client +acceptance remain open. Coordinate exchanged-token scope enforcement with the Web API before enabling +OAuth: messaging credentials must not inherit owner credential-management privileges. + Release-triggered npm publication, OIDC trusted publishing, provenance generation and version synchronization already exist in [publish.yml](.github/workflows/publish.yml). Preserve them. [CI](https://github.com/markmnl/fmsg-mcp/actions/workflows/tests.yml) already covers Node 22/24, diff --git a/docs/token-providers.md b/docs/token-providers.md new file mode 100644 index 0000000..508c01f --- /dev/null +++ b/docs/token-providers.md @@ -0,0 +1,93 @@ +# Web API token providers + +`FmsgClient` accepts either the existing `fmsgk_…` API-key string or a `TokenProvider`. +This is a client-library extension point. The MCP executable still uses API keys; +HTTP OAuth discovery, incoming-token validation and RFC 8693 exchange are separate work. + +Import `TokenProvider`, `TokenProviderRequest` and `AccessToken` from +`@markmnl/fmsg-mcp/client` (also exported from the package root): + +```ts +interface TokenProvider { + getToken(request: { + apiUrl: string; + signal: AbortSignal; + forceRefresh: boolean; + }): Promise<{ + accessToken: string; + address: string; + expiresAtMs: number; + }>; + close?(): void; +} +``` + +Supply an implementation as `new FmsgClient(apiUrl, tokenProvider, options)`. +One provider and client belong to one caller, authorization grant and Web API URL. +Sharing a provider between clients risks mixing grants or closing another client's credentials. + +## Provider responsibilities + +- Obtain a fresh **Web API** bearer access token whenever called. `apiUrl` is the + normalized, fixed upstream URL; never derive credential destinations from model input. + `forceRefresh: true` means the previous token was rejected with 401 or renewal was + explicitly requested. It must bypass any provider cache of that rejected token. +- Return the authenticated fmsg address that the token authorizes, preserving user-name + case, and a finite future expiry in milliseconds since the Unix epoch. Use trusted + authorization/exchange metadata; do not invent an expiry or infer identity from an + unverified incoming token. The generic client does not decode provider tokens. +- Honour `signal` for all network work. It aborts when renewal times out, the client + closes, or all callers waiting for that renewal cancel. Reject on failed/revoked + authorization; never fall back to an owner token, API key or another grant. +- Keep credentials out of errors and logs. `close()` synchronously releases retained + credentials/resources; it is called once when the owning client closes. Closing a + client is local cleanup, not remote OAuth-grant revocation. + +## Client responsibilities + +The client caches immutable token snapshots and shares concurrent renewal. It renews +five minutes before expiry by default, capped at half the remaining lifetime when the +token is acquired, so a one-minute token is usable without constant re-exchange. +`refreshMarginMs: 0` disables early renewal, but never permits an expired cached token. +Each acquisition has the client's `timeoutMs` budget (60 seconds by default). + +The first successful acquisition pins the address. A renewal for another address is +rejected before any request uses that token; changing identities requires a new client. +This is caller binding, not an additional messaging permission system. The Web API +still validates the actual credential, identity, scopes, visibility and host limits. + +Protected HTTP requests retry once on 401 after renewal. Concurrent or late 401s for +the same old token reuse an already renewed token; 403 never triggers renewal. +Failed renewal discards the cache. Provider output must be a bearer token with a valid +address and expiry; an API key passed back as an access token is rejected. + +`getToken(force?, signal?)` and `address(signal?)` support cancellation, as do existing +request methods that accept a signal. Cancelling one waiter leaves renewal available +to others; cancelling the last aborts acquisition. Late provider completion cannot +replace a later token. `close()` aborts outstanding token acquisition and HTTP work. + +`openFmsgWebSocket(client, signal?)` obtains its bearer from the same provider/cache. +The optional signal cancels token acquisition before opening the socket. The caller +owns the returned WebSocket and must handle its events and close it; long-lived +connections do not gain automatic token renewal or revocation handling from this helper. +`wait_for_message` already owns its socket and passes cancellation into acquisition. + +## Contract for a future OAuth adapter + +Validate the incoming MCP access token for the configured issuer and MCP audience, +then use authenticated [RFC 8693 token exchange](https://www.rfc-editor.org/rfc/rfc8693.html) +to obtain a separate upstream token. [MCP forbids passing the incoming token through +to the Web API](https://modelcontextprotocol.io/specification/2026-07-28/basic/authorization/security-considerations). +Keep issuer, JWKS/discovery URLs, audiences, address-claim mapping and client credentials +configurable; no particular identity provider is required by this interface. + +The IdP and Web API must agree on scopes and consented-identity binding before hosted +OAuth is enabled. Exchanged messaging tokens must not acquire owner key-management +rights or broaden identity via `X-FMSG-Act-As`. The Web API enforces those restrictions; +the tool list is not an authorization boundary. OAuth refresh/revocation and browser +consent are adapter/authorization-service work. Offline JWT validation alone does not +provide immediate revocation of already issued upstream tokens. + +Tests in `test/token-provider.test.ts` use registered opaque token fixtures to exercise +renewal, cancellation, caller isolation, and HTTP/WebSocket credential delivery. They +do not establish OAuth conformance, JWT verification, or compatibility with a real IdP. diff --git a/package.json b/package.json index 52f790f..5577850 100644 --- a/package.json +++ b/package.json @@ -46,6 +46,7 @@ "CHANGELOG.md", "ROADMAP.md", "docs/http-deployment.md", + "docs/token-providers.md", "LICENSE", "server.json" ], diff --git a/src/client/client.ts b/src/client/client.ts index 34e54c1..9c44394 100644 --- a/src/client/client.ts +++ b/src/client/client.ts @@ -3,6 +3,8 @@ import { normalizeMessageId, parseFmsgJson, stringifyWithIds } from "./message-i import { redactSecrets } from "./redact.js"; import { normalizeApiUrl } from "./url.js"; import { readBytes, withIdleTimeout } from "./stream.js"; +import { FmsgHttpError, readError } from "./errors.js"; +import { ApiKeyTokenProvider, type TokenProvider } from "./token-provider.js"; import type { AccessToken, Attachment, @@ -13,11 +15,13 @@ import type { Thread, } from "./types.js"; +export { FmsgHttpError } from "./errors.js"; + export type FetchLike = typeof fetch; export type FmsgClientOptions = { fetch?: FetchLike; - /** Refresh the access token this long before it expires (default 5 minutes). */ + /** Refresh margin (default 5 minutes), capped at half the token's remaining lifetime on acquisition. */ refreshMarginMs?: number; /** Per-request timeout; attachment streams use separate header/idle budgets (default 60 s). */ timeoutMs?: number; @@ -25,137 +29,122 @@ export type FmsgClientOptions = { allowInsecureHttp?: boolean; }; -/** An HTTP error from the fmsg Web API, with the status and the host's own error text. */ -export class FmsgHttpError extends Error { - constructor( - message: string, - readonly status: number, - readonly method: string, - readonly path: string, - /** Machine-readable `code` from the body, when the host sends one (thread routes). */ - readonly code?: string, - ) { - super(redactSecrets(message).text); - if (this.code) this.code = redactSecrets(this.code).text; - this.method = redactSecrets(method).text; - this.path = redactSecrets(path).text; - this.name = "FmsgHttpError"; - } -} - -function decodeJwtPayload(token: string): Record { - const parts = token.split("."); - if (parts.length !== 3 || !parts[1]) throw new Error("token exchange returned an invalid JWT"); - try { - return JSON.parse(Buffer.from(parts[1], "base64url").toString("utf8")) as Record; - } catch { - throw new Error("token exchange returned an unreadable JWT payload"); - } -} - -async function readError(response: Response): Promise<{ message: string; code?: string }> { - // Preserve canonical 400/413 JSON policy details. Other errors, including - // proxy pages, get a bounded preview independent of message acceptance limits. - const isJson = (response.headers.get("content-type") ?? "").toLowerCase().includes("json"); - let raw: string; - if (!isJson || ![400, 413].includes(response.status)) { - const { data, truncated } = await readBytes(response.body, 2048, true); - raw = Buffer.from(data).toString("utf8"); - if (!isJson || truncated) return { message: (raw || `HTTP ${response.status}`) + (truncated ? "\n[upstream response truncated at 2048 bytes]" : "") }; - } else raw = await response.text(); - if (!raw) return { message: `HTTP ${response.status}` }; - try { - const parsed = JSON.parse(raw) as { error?: unknown; code?: unknown }; - const message = typeof parsed.error === "string" ? parsed.error : `HTTP ${response.status}`; - return typeof parsed.code === "string" ? { message, code: parsed.code } : { message }; - } catch { - return { message: Buffer.byteLength(raw) > 2048 ? Buffer.from(raw).subarray(0, 2048).toString("utf8") + "\n[upstream response truncated at 2048 bytes]" : raw }; - } -} - function withId(message: FmsgMessage, id: string): FmsgMessage { return { ...message, id, terminal: message.terminal === true, reaction: message.reaction ?? null, reactions: message.reactions ?? [] }; } +type TokenRenewal = { promise: Promise; abort: AbortController; waiters: number }; + +/** Detach one waiter without cancelling work another caller still needs. */ +function withAbort(promise: Promise, signal?: AbortSignal): Promise { + if (!signal) return promise; + return new Promise((resolve, reject) => { + const onAbort = () => { signal.removeEventListener("abort", onAbort); reject(signal.reason); }; + signal.addEventListener("abort", onAbort, { once: true }); + if (signal.aborted) onAbort(); + // Always observe the operation, including a provider that ignores cancellation. + promise.then(resolve, reject).finally(() => signal.removeEventListener("abort", onAbort)); + }); +} + /** - * Client for the fmsg Web API (FMSG-003). Exchanges an `fmsgk_` API key for a - * short-lived JWT, refreshes it ahead of expiry, and retries once on 401. + * Client for the fmsg Web API (FMSG-003). Accepts an API key or a caller-bound + * token provider, renews tokens ahead of expiry, and retries once on 401. */ export class FmsgClient { readonly apiUrl: string; private token?: AccessToken; - private tokenPromise?: Promise; + private renewal?: TokenRenewal; + private provider?: TokenProvider; + private boundAddress?: string; + private refreshAtMs = 0; private readonly lifetime = new AbortController(); constructor( apiUrl: string, - private apiKey: string, + credentials: string | TokenProvider, private readonly options: FmsgClientOptions = {}, ) { this.apiUrl = normalizeApiUrl(apiUrl, options.allowInsecureHttp); - if (!apiKey.startsWith("fmsgk_")) throw new Error("fmsg API key must start with fmsgk_"); + if (options.refreshMarginMs !== undefined && (!Number.isFinite(options.refreshMarginMs) || options.refreshMarginMs < 0)) { + throw new Error("refreshMarginMs must be a finite non-negative number"); + } + this.provider = typeof credentials === "string" ? new ApiKeyTokenProvider(credentials, this.fetchImpl) : credentials; + if (!this.provider || typeof this.provider.getToken !== "function") throw new Error("a token provider must implement getToken"); } private get fetchImpl(): FetchLike { return this.options.fetch ?? fetch; } - /** The address this client acts as (from the JWT `sub`), exchanging the key if needed. */ - async address(): Promise { - return (await this.getToken()).address; + /** The provider's authenticated address, pinned for this client's lifetime. */ + async address(signal?: AbortSignal): Promise { + return (await this.getToken(false, signal)).address; } - async getToken(force = false): Promise { + async getToken(force = false, signal?: AbortSignal): Promise { + signal?.throwIfAborted(); this.lifetime.signal.throwIfAborted(); - const margin = this.options.refreshMarginMs ?? 300_000; - if (this.tokenPromise) return this.tokenPromise; - if (!force && this.token && this.token.expiresAtMs - margin > Date.now()) return this.token; - this.tokenPromise = this.exchangeToken(); + if (!this.renewal && !force && this.token && this.refreshAtMs > Date.now()) return this.token; + const renewal = this.renewal ?? this.startRenewal(force); + renewal.waiters++; try { - this.token = await this.tokenPromise; - this.lifetime.signal.throwIfAborted(); - return this.token; - } catch (error) { - this.token = undefined; - throw error; + return await withAbort(renewal.promise, signal); } finally { - this.tokenPromise = undefined; + renewal.waiters--; + if (!renewal.waiters && this.renewal === renewal) { + this.renewal = undefined; + renewal.abort.abort(); + } } } + private startRenewal(forceRefresh: boolean): TokenRenewal { + const provider = this.provider!; + const renewal: TokenRenewal = { promise: undefined!, abort: new AbortController(), waiters: 0 }; + this.renewal = renewal; + this.token = undefined; + const signal = AbortSignal.any([this.lifetime.signal, renewal.abort.signal]); + const timer = setTimeout(() => renewal.abort.abort(new DOMException("token renewal timed out", "TimeoutError")), this.options.timeoutMs ?? 60_000).unref(); + const acquiring = Promise.resolve().then(() => { + signal.throwIfAborted(); + return provider.getToken({ apiUrl: this.apiUrl, signal, forceRefresh }); + }); + renewal.promise = withAbort(acquiring, signal).then(value => { + signal.throwIfAborted(); + const address = typeof value?.address === "string" ? normalizeFmsgAddress(value.address) : undefined; + if (!address) throw new Error("token provider returned an invalid fmsg address"); + if (typeof value.accessToken !== "string" || !/^[A-Za-z0-9._~+\/-]+=*$/u.test(value.accessToken) || value.accessToken.startsWith("fmsgk_")) { + throw new Error("token provider must return a Web API bearer access token"); + } + const now = Date.now(); + if (!Number.isFinite(value.expiresAtMs) || value.expiresAtMs <= now) throw new Error("token provider returned an invalid or expired token lifetime"); + if (this.boundAddress !== undefined && address !== this.boundAddress) throw new Error("token provider changed the authenticated address; create a new client for a different identity"); + this.boundAddress = address; + const token = Object.freeze({ accessToken: value.accessToken, address, expiresAtMs: value.expiresAtMs }); + this.refreshAtMs = token.expiresAtMs - Math.min(this.options.refreshMarginMs ?? 300_000, (token.expiresAtMs - now) / 2); + this.token = token; + return token; + }).finally(() => { + clearTimeout(timer); + if (this.renewal === renewal) this.renewal = undefined; + }); + return renewal; + } + /** Release credentials and cancel outstanding work when the client is no longer used. */ close(): void { + if (this.lifetime.signal.aborted) return; this.lifetime.abort(); - this.apiKey = ""; this.token = undefined; - } - - private async exchangeToken(): Promise { - const response = await this.fetchImpl(`${this.apiUrl}/fmsg/token`, { - method: "POST", - headers: { authorization: `Bearer ${this.apiKey}` }, - redirect: "error", - signal: AbortSignal.any([this.lifetime.signal, AbortSignal.timeout(this.options.timeoutMs ?? 60_000)]), - }); - if (!response.ok) { - const { message, code } = await readError(response); - throw new FmsgHttpError(`token exchange failed: ${message}`, response.status, "POST", "/fmsg/token", code); - } - const body = (await response.json()) as { access_token?: unknown; expires_in?: unknown; expires_at?: unknown }; - if (typeof body.access_token !== "string") throw new Error("token response has no access_token"); - const payload = decodeJwtPayload(body.access_token); - const address = typeof payload.sub === "string" ? normalizeFmsgAddress(payload.sub) : undefined; - if (!address) throw new Error("token JWT sub is not an fmsg address"); - const fromResponse = typeof body.expires_at === "string" ? Date.parse(body.expires_at) : Number.NaN; - const fromJwt = typeof payload.exp === "number" ? payload.exp * 1000 : Number.NaN; - const fromIn = typeof body.expires_in === "number" ? Date.now() + body.expires_in * 1000 : Number.NaN; - const expiresAtMs = [fromResponse, fromJwt, fromIn].find(Number.isFinite) ?? Date.now() + 3_600_000; - return { accessToken: body.access_token, address, expiresAtMs }; + const provider = this.provider; + this.provider = undefined; + provider?.close?.(); } private async request(path: string, init: RequestInit = {}, retry401 = true, streaming = false): Promise { init.signal?.throwIfAborted(); - const token = await this.getToken(); + const token = await this.getToken(false, init.signal ?? undefined); const headers = new Headers(init.headers); headers.set("authorization", `Bearer ${token.accessToken}`); const headerDeadline = new AbortController(); @@ -167,7 +156,10 @@ export class FmsgClient { const response = await this.fetchImpl(`${this.apiUrl}${path}`, { ...init, headers, signal, redirect: "error" }); if (response.status === 401 && retry401) { await response.body?.cancel(); - await this.getToken(true); + // A slower 401 may refer to a token another request already renewed. + if (!this.token || this.token === token || this.token.expiresAtMs <= Date.now()) { + await this.getToken(true, init.signal ?? undefined); + } return this.request(path, init, false, streaming); } if (!response.ok) { @@ -376,7 +368,7 @@ export class FmsgClient { async send(input: SendInput): Promise { if (input.to.length === 0) throw new Error("at least one recipient is required"); if (input.pid && input.topic) throw new Error("a reply (pid) cannot carry a topic"); - const from = await this.address(); + const from = await this.address(input.signal); const body = redactSecrets(input.body); const topic = redactSecrets(input.pid ? "" : (input.topic ?? "")); const draftId = await this.createDraft({ ...input, body: body.text, topic: topic.text }, from); diff --git a/src/client/errors.ts b/src/client/errors.ts new file mode 100644 index 0000000..bfeafde --- /dev/null +++ b/src/client/errors.ts @@ -0,0 +1,40 @@ +import { redactSecrets } from "./redact.js"; +import { readBytes } from "./stream.js"; + +/** An HTTP error from the fmsg Web API, with the status and the host's own error text. */ +export class FmsgHttpError extends Error { + constructor( + message: string, + readonly status: number, + readonly method: string, + readonly path: string, + /** Machine-readable `code` from the body, when the host sends one (thread routes). */ + readonly code?: string, + ) { + super(redactSecrets(message).text); + if (this.code) this.code = redactSecrets(this.code).text; + this.method = redactSecrets(method).text; + this.path = redactSecrets(path).text; + this.name = "FmsgHttpError"; + } +} + +export async function readError(response: Response): Promise<{ message: string; code?: string }> { + // Preserve canonical 400/413 JSON policy details. Other errors, including + // proxy pages, get a bounded preview independent of message acceptance limits. + const isJson = (response.headers.get("content-type") ?? "").toLowerCase().includes("json"); + let raw: string; + if (!isJson || ![400, 413].includes(response.status)) { + const { data, truncated } = await readBytes(response.body, 2048, true); + raw = Buffer.from(data).toString("utf8"); + if (!isJson || truncated) return { message: (raw || `HTTP ${response.status}`) + (truncated ? "\n[upstream response truncated at 2048 bytes]" : "") }; + } else raw = await response.text(); + if (!raw) return { message: `HTTP ${response.status}` }; + try { + const parsed = JSON.parse(raw) as { error?: unknown; code?: unknown }; + const message = typeof parsed.error === "string" ? parsed.error : `HTTP ${response.status}`; + return typeof parsed.code === "string" ? { message, code: parsed.code } : { message }; + } catch { + return { message: Buffer.byteLength(raw) > 2048 ? Buffer.from(raw).subarray(0, 2048).toString("utf8") + "\n[upstream response truncated at 2048 bytes]" : raw }; + } +} diff --git a/src/client/index.ts b/src/client/index.ts index 9d78097..0985139 100644 --- a/src/client/index.ts +++ b/src/client/index.ts @@ -1,4 +1,5 @@ export { FmsgClient, FmsgHttpError, type FmsgClientOptions } from "./client.js"; +export type { TokenProvider, TokenProviderRequest } from "./token-provider.js"; export { openFmsgWebSocket, parseWsEvent } from "./ws.js"; export { redactSecrets, safeErrorMessage, type Redacted } from "./redact.js"; export { normalizeMessageId, compareMessageIds, maxMessageId, parseFmsgJson, stringifyWithIds } from "./message-id.js"; diff --git a/src/client/token-provider.ts b/src/client/token-provider.ts new file mode 100644 index 0000000..58a1533 --- /dev/null +++ b/src/client/token-provider.ts @@ -0,0 +1,69 @@ +import { normalizeFmsgAddress } from "../address.js"; +import { FmsgHttpError, readError } from "./errors.js"; +import type { AccessToken } from "./types.js"; + +export type TokenProviderRequest = { + /** Normalized Web API URL this client is bound to. */ + readonly apiUrl: string; + /** Aborted on timeout, client close, or cancellation by all waiting callers. */ + readonly signal: AbortSignal; + /** A protected request rejected the previous token, or renewal was explicitly forced. */ + readonly forceRefresh: boolean; +}; + +/** + * Supplies a fresh Web API access token for one caller and authorization grant. + * FmsgClient owns caching and concurrent renewal. Never return an incoming MCP + * token or use a provider across different callers/grants. See docs/token-providers.md. + */ +export interface TokenProvider { + getToken(request: TokenProviderRequest): Promise; + /** Release credentials/resources. Called once by FmsgClient.close(). */ + close?(): void; +} + +/** Existing API-key exchange, behind the same interface as future OAuth adapters. */ +export class ApiKeyTokenProvider implements TokenProvider { + constructor(private apiKey: string, private readonly fetchImpl: typeof fetch) { + if (!apiKey.startsWith("fmsgk_")) throw new Error("fmsg API key must start with fmsgk_"); + } + + async getToken({ apiUrl, signal }: TokenProviderRequest): Promise { + const response = await this.fetchImpl(`${apiUrl}/fmsg/token`, { + method: "POST", headers: { authorization: `Bearer ${this.apiKey}` }, redirect: "error", signal, + }); + if (!response.ok) { + const { message, code } = await readError(response); + throw new FmsgHttpError(`token exchange failed: ${message}`, response.status, "POST", "/fmsg/token", code); + } + const body = (await response.json()) as { access_token?: unknown; expires_in?: unknown; expires_at?: unknown }; + if (typeof body.access_token !== "string") throw new Error("token response has no access_token"); + const payload = decodeJwtPayload(body.access_token); + const address = typeof payload.sub === "string" ? normalizeFmsgAddress(payload.sub) : undefined; + if (!address) throw new Error("token JWT sub is not an fmsg address"); + const expiries = [ + typeof body.expires_at === "string" ? Date.parse(body.expires_at) : Number.NaN, + typeof payload.exp === "number" ? payload.exp * 1000 : Number.NaN, + typeof body.expires_in === "number" ? Date.now() + body.expires_in * 1000 : Number.NaN, + ].filter(Number.isFinite); + // Never extend the JWT lifetime using a later response expiry or a guessed TTL. + const expiresAtMs = expiries.length ? Math.min(...expiries) : Number.NaN; + return { accessToken: body.access_token, address, expiresAtMs }; + } + + close(): void { this.apiKey = ""; } +} + +// Only the first-party API-key contract requires sub/exp in a JWT. Custom +// providers supply address/expiry metadata; the client does not decode their tokens. +function decodeJwtPayload(token: string): Record { + const parts = token.split("."); + if (parts.length !== 3 || !parts[1]) throw new Error("token exchange returned an invalid JWT"); + try { + const payload: unknown = JSON.parse(Buffer.from(parts[1], "base64url").toString("utf8")); + if (!payload || typeof payload !== "object" || Array.isArray(payload)) throw new Error(); + return payload as Record; + } catch { + throw new Error("token exchange returned an unreadable JWT payload"); + } +} diff --git a/src/client/types.ts b/src/client/types.ts index f3cd38f..9401951 100644 --- a/src/client/types.ts +++ b/src/client/types.ts @@ -99,10 +99,10 @@ export type Thread = { }; export type AccessToken = { - accessToken: string; - /** fmsg address from the JWT `sub` claim. */ - address: string; - expiresAtMs: number; + readonly accessToken: string; + /** Authenticated fmsg address supplied by the provider; not necessarily a JWT sub. */ + readonly address: string; + readonly expiresAtMs: number; }; export type OutboundAttachment = { diff --git a/src/client/ws.ts b/src/client/ws.ts index 269cad2..e5c2aa3 100644 --- a/src/client/ws.ts +++ b/src/client/ws.ts @@ -4,8 +4,9 @@ import { normalizeMessageId, parseFmsgJson } from "./message-id.js"; import type { FmsgMessage, WsEvent } from "./types.js"; /** Open the event WebSocket, authenticating with the bearer JWT in the header. */ -export async function openFmsgWebSocket(client: FmsgClient): Promise { - const token = await client.getToken(); +export async function openFmsgWebSocket(client: FmsgClient, signal?: AbortSignal): Promise { + const token = await client.getToken(false, signal); + signal?.throwIfAborted(); const url = new URL(client.apiUrl); url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; url.pathname = `${url.pathname.replace(/\/+$/u, "")}/fmsg/ws`; diff --git a/src/context.ts b/src/context.ts index cecc9be..e4a07c0 100644 --- a/src/context.ts +++ b/src/context.ts @@ -1,11 +1,11 @@ import type { AuthInfo, ServerContext } from "@modelcontextprotocol/server"; import { FmsgClient } from "./client/client.js"; -/** A resolved caller: the client bound to one API key and the address it acts as. */ +/** A resolved caller: the client bound to one authorization grant and the address it acts as. */ export type Caller = { client: FmsgClient; address: string; - /** When the key's exchanged token expires (ms since epoch), for whoami. */ + /** When the current Web API access token expires (ms since epoch), for whoami. */ tokenExpiresAt: () => Promise; }; diff --git a/src/wait.ts b/src/wait.ts index b905a4b..641078c 100644 --- a/src/wait.ts +++ b/src/wait.ts @@ -41,7 +41,7 @@ export type WaitResult = { note: string | null; }; -type Deps = { openSocket?: (client: FmsgClient) => Promise }; +type Deps = { openSocket?: (client: FmsgClient, signal?: AbortSignal) => Promise }; /** * Block until the next qualifying inbound message (plus any that arrive on the @@ -285,7 +285,7 @@ export async function waitForMessage( }; const open = deps.openSocket ?? openFmsgWebSocket; - open(client) + open(client, retrySignal) .then((ws) => { if (finished) { ws.close(); diff --git a/test/fake-fmsg-server.ts b/test/fake-fmsg-server.ts index f2a2955..6ee766a 100644 --- a/test/fake-fmsg-server.ts +++ b/test/fake-fmsg-server.ts @@ -7,6 +7,7 @@ import { createServer, type IncomingMessage, type ServerResponse } from "node:ht import { createHash } from "node:crypto"; import type { AddressInfo } from "node:net"; import { WebSocket, WebSocketServer } from "ws"; +import type { AccessToken } from "../src/client/types.js"; export type StoredMessage = { id: string; @@ -80,6 +81,8 @@ export class FakeFmsgServer { ["fmsgk_agent_secret", "@Alice_ChatGPT@example.com"], ]); private readonly tokenKeys = new Map(); + /** Registered provider fixtures only; this does not simulate JWT verification or OAuth. */ + readonly providerTokens = new Map(); /** Fail the next request whose path matches, with this status and message. */ failNext: { match: RegExp; status: number; error: string; code?: string } | undefined; /** Force the next protected request to answer 401 (expired JWT simulation). */ @@ -255,6 +258,8 @@ export class FakeFmsgServer { } private authenticatedSubject(token: string | undefined): string | undefined { + const provided = token ? this.providerTokens.get(token) : undefined; + if (provided) return provided.expiresAtMs > Date.now() ? provided.address : undefined; const subject = subjectOf(token); const key = token ? this.tokenKeys.get(token) : undefined; return subject && key && this.apiKeys.get(key) === subject ? subject : undefined; diff --git a/test/http.test.ts b/test/http.test.ts index 2c856ea..b7e50c1 100644 --- a/test/http.test.ts +++ b/test/http.test.ts @@ -57,9 +57,11 @@ describe("HTTP transport", () => { expect(notKey.status).toBe(401); }); - it("releases the caller lease when middleware rejects an expired upstream token", async () => { - fake.tokenTtlSeconds = -1; - const verify = vi.spyOn(http.provider, "verifyAccessToken"); + it("releases the caller lease when middleware rejects expired authentication", async () => { + const originalVerify = http.provider.verifyAccessToken.bind(http.provider); + const verify = vi.spyOn(http.provider, "verifyAccessToken").mockImplementation(async token => ({ + ...await originalVerify(token), expiresAt: Math.floor(Date.now() / 1000) - 1, + })); try { const response = await fetch(`${base}/mcp`, { method: "POST", diff --git a/test/token-provider.test.ts b/test/token-provider.test.ts new file mode 100644 index 0000000..0189ff9 --- /dev/null +++ b/test/token-provider.test.ts @@ -0,0 +1,269 @@ +import { once } from "node:events"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { FmsgClient, openFmsgWebSocket, type AccessToken, type TokenProvider, type TokenProviderRequest } from "../src/client/index.js"; +import { FakeFmsgServer } from "./fake-fmsg-server.js"; +import { ALICE, BOB } from "./helpers.js"; + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +} + +describe("caller-bound token providers", () => { + let fake: FakeFmsgServer; + let clients: FmsgClient[]; + let nextToken: number; + beforeEach(async () => { + fake = new FakeFmsgServer(); + await fake.start(); + clients = []; + nextToken = 0; + }); + afterEach(async () => { + clients.forEach(client => client.close()); + vi.useRealTimers(); + vi.restoreAllMocks(); + await fake.stop(); + }); + + function issue(address = ALICE, ttlMs = 3_600_000): AccessToken { + const token = { accessToken: `fixture-token-${++nextToken}`, address, expiresAtMs: Date.now() + ttlMs }; + fake.providerTokens.set(token.accessToken, token); + return token; + } + + function provider(address = ALICE, ttlMs = 3_600_000) { + return { getToken: vi.fn(async (_request: TokenProviderRequest) => issue(address, ttlMs)), close: vi.fn() } satisfies TokenProvider; + } + + function client(credentials: string | TokenProvider, options: ConstructorParameters[2] = {}) { + const result = new FmsgClient(fake.baseUrl, credentials, options); + clients.push(result); + return result; + } + + it("authenticates HTTP and WebSockets with provider tokens and isolates two callers", async () => { + const alice = client(provider()); + const bob = client(provider(BOB)); + const privateMessage = fake.seed({ from: BOB, to: [ALICE], data: "alice inbox" }); + const [aliceInbox, bobInbox] = await Promise.all([alice.listInbox(), bob.listInbox()]); + expect(aliceInbox.map(m => m.id)).toEqual([privateMessage.id]); + expect(bobInbox).toEqual([]); + const secret = fake.seed({ from: ALICE, to: [ALICE], data: "alice only" }); + await expect(bob.getMessage(secret.id)).rejects.toMatchObject({ status: 404 }); + const sent = await alice.send({ to: [BOB], body: "hello" }); + expect(fake.messages.get(sent.id)?.from).toBe(ALICE); + const a = await openFmsgWebSocket(alice); + const b = await openFmsgWebSocket(bob); + try { + await Promise.all([once(a, "open"), once(b, "open")]); + expect(fake.connectedSockets(ALICE)).toBe(1); + expect(fake.connectedSockets(BOB)).toBe(1); + const received = once(b, "message"); + fake.push(fake.seed({ from: ALICE, to: [BOB], data: "bob event" })); + expect(String((await received)[0])).toContain("bob event"); + } finally { a.close(); b.close(); } + expect(fake.requests.some(r => r.path === "/fmsg/token")).toBe(false); + }); + + it("renews short-lived tokens once per batch without refreshing on every call", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const source = provider(ALICE, 60_000); + const c = client(source); + const first = await c.getToken(); + await c.listInbox(); + await c.address(); + expect(source.getToken).toHaveBeenCalledTimes(1); + vi.setSystemTime(Date.now() + 30_001); + const refreshed = await Promise.all(Array.from({ length: 10 }, () => c.getToken())); + expect(source.getToken).toHaveBeenCalledTimes(2); + expect(new Set(refreshed.map(t => t.accessToken)).size).toBe(1); + expect(refreshed[0]!.accessToken).not.toBe(first.accessToken); + expect(source.getToken.mock.calls[1]![0]).toMatchObject({ apiUrl: fake.baseUrl, forceRefresh: false }); + vi.setSystemTime(Date.now() + 60_001); + await c.listInbox(); + expect(source.getToken).toHaveBeenCalledTimes(3); + }); + + it("keeps API-key renewal ahead of expiry and does not extend JWT expiry from response metadata", async () => { + vi.useFakeTimers({ toFake: ["Date"] }); + const c = client("fmsgk_alice_secret"); + const first = await c.getToken(); + vi.setSystemTime(first.expiresAtMs - 300_001); + await c.getToken(); + expect(fake.requests.filter(r => r.path === "/fmsg/token")).toHaveLength(1); + vi.setSystemTime(first.expiresAtMs - 299_999); + await c.getToken(); + expect(fake.requests.filter(r => r.path === "/fmsg/token")).toHaveLength(2); + const payload = Buffer.from(JSON.stringify({ sub: ALICE, exp: Math.floor(Date.now() / 1000) + 60 })).toString("base64url"); + const bounded = client("fmsgk_alice_secret", { fetch: async () => Response.json({ + access_token: `e30.${payload}.signature`, expires_at: new Date(Date.now() + 3_600_000).toISOString(), expires_in: 3600, + }) }); + expect((await bounded.getToken()).expiresAtMs).toBe((Math.floor(Date.now() / 1000) + 60) * 1000); + }); + + it("shares forced renewal and reuses it when an older request returns a late 401", async () => { + const source = provider(); + const oldResponse = deferred(); + const requests: string[] = []; + let oldToken: string; + const c = client(source, { fetch: async (_url, init) => { + const bearer = new Headers(init?.headers).get("authorization")!; + requests.push(bearer); + if (bearer === `Bearer ${oldToken}`) { + return requests.length === 1 ? oldResponse.promise : Response.json({ error: "expired" }, { status: 401 }); + } + return Response.json([]); + } }); + oldToken = (await c.getToken()).accessToken; + const slow = c.listInbox(); + const fast = Array.from({ length: 5 }, () => c.listInbox()); + await Promise.all(fast); + expect(source.getToken).toHaveBeenCalledTimes(2); + expect(source.getToken.mock.calls[1]![0].forceRefresh).toBe(true); + oldResponse.resolve(Response.json({ error: "old token" }, { status: 401 })); + await slow; + expect(source.getToken).toHaveBeenCalledTimes(2); + expect(requests).toHaveLength(12); + }); + + it("retries a protected 401 only once, and never renews for an upstream 403", async () => { + const source = provider(); + let status = 401; + const fetch = vi.fn(async () => Response.json({ error: "upstream denial" }, { status })); + const c = client(source, { fetch }); + await expect(c.listInbox()).rejects.toMatchObject({ status: 401, message: "upstream denial" }); + expect(fetch).toHaveBeenCalledTimes(2); + expect(source.getToken).toHaveBeenCalledTimes(2); + status = 403; + await expect(c.listInbox()).rejects.toMatchObject({ status: 403 }); + expect(source.getToken).toHaveBeenCalledTimes(2); + }); + + it("does not fall back to a cached token after renewal fails, and can recover", async () => { + const source = provider(); + const c = client(source); + await c.getToken(); + source.getToken.mockRejectedValueOnce(new Error("connection revoked")); + fake.rejectNextProtected = true; + await expect(c.listInbox()).rejects.toThrow("connection revoked"); + expect(fake.requests.filter(r => r.path === "/fmsg")).toHaveLength(1); + await expect(c.listInbox()).resolves.toEqual([]); + expect(source.getToken).toHaveBeenCalledTimes(3); + }); + + it("pins the address through failed renewals and snapshots provider output", async () => { + const supplied = { ...issue() }; + const source = provider(); + source.getToken.mockResolvedValueOnce(supplied); + const c = client(source); + const first = await c.getToken(); + supplied.address = BOB; + supplied.accessToken = "changed"; + expect(first.address).toBe(ALICE); + expect(first.accessToken).not.toBe("changed"); + expect(Object.isFrozen(first)).toBe(true); + source.getToken.mockResolvedValueOnce(issue(BOB)); + await expect(c.getToken(true)).rejects.toThrow("changed the authenticated address"); + source.getToken.mockResolvedValueOnce(issue("@Alice@example.com")); + await expect(c.listInbox()).rejects.toThrow("changed the authenticated address"); + expect(fake.requests).toHaveLength(0); + await expect(c.address()).resolves.toBe(ALICE); + }); + + it.each([ + { accessToken: "" }, { accessToken: "secret\r\ninjection" }, { accessToken: "fmsgk_wrong_credential" }, + { address: "invalid" }, { expiresAtMs: Number.NaN }, { expiresAtMs: Number.POSITIVE_INFINITY }, { expiresAtMs: 0 }, + ])("rejects invalid provider output before contacting the Web API: %j", async override => { + const c = client({ getToken: async () => ({ ...issue(), ...override }) }); + await expect(c.listInbox()).rejects.toThrow(/token provider/u); + expect(fake.requests).toHaveLength(0); + }); + + it("cancels one token waiter without cancelling another, then aborts when no callers remain", async () => { + const source = provider(); + const held = deferred(); + source.getToken.mockImplementationOnce(() => held.promise); + const c = client(source); + const a = new AbortController(); + const b = new AbortController(); + const one = expect(c.listInbox(20, 0, a.signal)).rejects.toMatchObject({ name: "AbortError" }); + const two = expect(c.listInbox(20, 0, b.signal)).rejects.toMatchObject({ name: "AbortError" }); + await vi.waitFor(() => expect(source.getToken).toHaveBeenCalledOnce()); + const renewalSignal = source.getToken.mock.calls[0]![0].signal; + a.abort(); + await one; + expect(renewalSignal.aborted).toBe(false); + b.abort(); + await two; + expect(renewalSignal.aborted).toBe(true); + const fresh = await c.getToken(); + held.resolve(issue(BOB)); // Late completion must not replace the recovered token. + await Promise.resolve(); + expect((await c.getToken()).accessToken).toBe(fresh.accessToken); + expect(fake.requests).toHaveLength(0); + }); + + it("lets an uncancelled waiter complete a shared renewal", async () => { + const source = provider(); + const held = deferred(); + source.getToken.mockImplementationOnce(() => held.promise); + const c = client(source); + const abort = new AbortController(); + const cancelled = expect(c.getToken(false, abort.signal)).rejects.toMatchObject({ name: "AbortError" }); + const remaining = c.listInbox(); + await vi.waitFor(() => expect(source.getToken).toHaveBeenCalledOnce()); + abort.abort(); + await cancelled; + held.resolve(issue()); + await expect(remaining).resolves.toEqual([]); + expect(source.getToken).toHaveBeenCalledOnce(); + }); + + it.each(["close", "timeout"])("settles waiting callers on %s even if a provider ignores its signal", async reason => { + vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const source = provider(); + const held = deferred(); + source.getToken.mockImplementationOnce(() => held.promise); + const timerSpy = vi.spyOn(globalThis, "setTimeout"); + const clearSpy = vi.spyOn(globalThis, "clearTimeout"); + const c = client(source, { timeoutMs: 100 }); + const pending = expect(c.getToken()).rejects.toMatchObject({ name: reason === "close" ? "AbortError" : "TimeoutError" }); + await vi.advanceTimersByTimeAsync(0); + if (reason === "close") c.close(); + else await vi.advanceTimersByTimeAsync(101); + await pending; + expect(source.getToken.mock.calls[0]![0].signal.aborted).toBe(true); + // Other tests may leave fetch keep-alive timers; check this acquisition timer. + expect(clearSpy).toHaveBeenCalledWith(timerSpy.mock.results[0]!.value); + c.close(); + c.close(); + expect(source.close).toHaveBeenCalledOnce(); + held.resolve(issue()); + await expect(c.getToken()).rejects.toMatchObject({ name: "AbortError" }); + }); + + it("does not start token acquisition for cancelled sends or WebSockets", async () => { + const source = provider(); + const c = client(source); + await expect(c.send({ to: [BOB], body: "cancelled", signal: AbortSignal.abort() })).rejects.toMatchObject({ name: "AbortError" }); + await expect(openFmsgWebSocket(c, AbortSignal.abort())).rejects.toMatchObject({ name: "AbortError" }); + expect(source.getToken).not.toHaveBeenCalled(); + expect(fake.requests).toHaveLength(0); + }); + + it("cancels token acquisition before opening a WebSocket", async () => { + const source = provider(); + source.getToken.mockImplementationOnce(() => new Promise(() => undefined)); + const c = client(source); + const abort = new AbortController(); + const opening = expect(openFmsgWebSocket(c, abort.signal)).rejects.toMatchObject({ name: "AbortError" }); + await vi.waitFor(() => expect(source.getToken).toHaveBeenCalledOnce()); + abort.abort(); + await opening; + expect(source.getToken.mock.calls[0]![0].signal.aborted).toBe(true); + expect(fake.connectedSockets(ALICE)).toBe(0); + }); +});