From f83b3401aced68eb75ccbe96887e6d76a9116f5f Mon Sep 17 00:00:00 2001 From: Sam Curry Date: Thu, 15 Oct 2026 17:31:54 +0800 Subject: [PATCH 1/5] feat(subscriptions): closeOnTokenExpiry and keepAlive options for useSubscriptionsServer closeOnTokenExpiry closes a connection with 4403 Forbidden when the exp of the token it connected with passes, which ends the subscriptions running on it. Until now an expired token was refused only on the next subscribe, and running subscriptions continued. keepAlive sets graphql-ws's ping interval. --- README.md | 4 + package-lock.json | 4 +- package.json | 2 +- src/subscriptions/server.spec.ts | 121 ++++++++++++++++++++++++++++++- src/subscriptions/server.ts | 53 +++++++++++++- 5 files changed, 179 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 23f6c4e..ff07430 100644 --- a/README.md +++ b/README.md @@ -382,6 +382,10 @@ This library includes a `subscriptions` module to provide simple setup using the Pass `includeDeprecatedElements: true` to add [deprecation usage](#collectdeprecatedelementusage) to the operation log entry as `deprecatedElements`. It is collected once, when the subscription is established, rather than per emitted payload — the usage is a property of the operation, not of each event. + Pass `closeOnTokenExpiry: true` to close a connection with `4403 Forbidden` when the `exp` of the token it connected with passes, which ends every subscription running on it. Without it, an expired token is refused only when the client next subscribes on the connection, and subscriptions already running continue. graphql-ws clients retry `4403`, so a client reconnects, and `connectionParams` (or whatever authorises the upgrade request) supplies a fresh token. + + `keepAlive` sets the milliseconds between the server's websocket pings (default 12 seconds, graphql-ws's default; `0` turns them off). A connection that has not answered a ping by the time the next one is due is terminated. The pings also stop proxies and load balancers from closing an idle connection, so keep the interval below their idle timeouts. + Example for Apollo Server (`wsServerCleanup` called in the `drainServer` plugin callback): ```ts diff --git a/package-lock.json b/package-lock.json index 150cc77..f3737a7 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@makerx/graphql-core", - "version": "3.4.0", + "version": "3.5.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@makerx/graphql-core", - "version": "3.4.0", + "version": "3.5.0", "license": "MIT", "devDependencies": { "@arethetypeswrong/cli": "^0.18.5", diff --git a/package.json b/package.json index 48ba42c..bc63c70 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@makerx/graphql-core", - "version": "3.4.0", + "version": "3.5.0", "private": false, "description": "A set of core GraphQL utilities that MakerX uses to build GraphQL APIs", "author": "MakerX", diff --git a/src/subscriptions/server.spec.ts b/src/subscriptions/server.spec.ts index 2f388e2..3f193b9 100644 --- a/src/subscriptions/server.spec.ts +++ b/src/subscriptions/server.spec.ts @@ -1,8 +1,9 @@ import type { Logger } from '@makerx/node-common' -import { GraphQLBoolean, GraphQLObjectType, GraphQLSchema, GraphQLString } from 'graphql' +import { GraphQLBoolean, GraphQLInt, GraphQLObjectType, GraphQLSchema, GraphQLString } from 'graphql' import { createClient, type Client } from 'graphql-ws' import { createServer, type Server } from 'http' import type { AddressInfo } from 'net' +import { setTimeout as delay } from 'timers/promises' import { afterEach, describe, expect, it, vi } from 'vitest' import WebSocket from 'ws' import type { GraphQLContext, JwtPayload } from '../context' @@ -20,6 +21,15 @@ const schema = new GraphQLSchema({ yield { token: context.user?.token } }, }, + ticks: { + type: GraphQLInt, + subscribe: async function* () { + for (let tick = 0; ; tick++) { + yield { ticks: tick } + await delay(20) + } + }, + }, }, }), }) @@ -79,6 +89,41 @@ const subscribeToToken = (client: Client) => client.subscribe({ query: 'subscription { token }' }, { next: ({ data }) => resolve(data?.token), error: reject, complete: () => {} }) }) +const closeCodeOf = (error: unknown) => + typeof error === 'object' && error !== null && 'code' in error && typeof error.code === 'number' ? error.code : undefined + +/** Runs a `ticks` subscription for up to `ms`, and reports how many ticks arrived and the close code that ended it, if any. */ +const watchTicks = (client: Client, ms: number) => + new Promise<{ ticks: number; closeCode?: number }>((resolve) => { + let ticks = 0 + const unsubscribe = client.subscribe( + { query: 'subscription { ticks }' }, + { + next: () => { + ticks++ + }, + error: (error) => resolve({ ticks, closeCode: closeCodeOf(error) }), + complete: () => resolve({ ticks }), + }, + ) + void delay(ms).then(() => { + unsubscribe() + resolve({ ticks }) + }) + }) + +const epochSeconds = (offsetSeconds: number) => Math.floor(Date.now() / 1000) + offsetSeconds + +const pingsWithin = async (url: string, ms: number) => { + const socket = new WebSocket(url, 'graphql-transport-ws') + cleanup.push(() => socket.close()) + let pings = 0 + socket.on('ping', () => pings++) + await new Promise((resolve) => socket.once('open', resolve)) + await delay(ms) + return pings +} + describe('useSubscriptionsServer', () => { it('verifies the bearer token from the connection params by default', async () => { const verifyToken = vi.fn(async () => claims) @@ -118,4 +163,78 @@ describe('useSubscriptionsServer', () => { expect(verifyToken).not.toHaveBeenCalled() expect(logger.error).toHaveBeenCalledWith('No auth token was supplied with the websocket connection') }) + + describe('closeOnTokenExpiry', () => { + const bearer = { connectionParams: { Authorization: 'Bearer token-params' } } + + it('closes the connection with 4403 when the token expires, ending a running subscription', async () => { + const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(1) })) + const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + + const result = await watchTicks(connect(url, bearer), 3000) + + expect(result.closeCode).toBe(4403) + expect(result.ticks).toBeGreaterThan(0) + expect(logger.info).toHaveBeenCalledWith('Subscription connection auth token expired', { + claims: { oid: 'oid-1', iss: 'https://issuer.example' }, + }) + }) + + it('closes the connection at once when the token has already expired', async () => { + const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(-10) })) + const { url } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + + const result = await watchTicks(connect(url, bearer), 1000) + + expect(result.closeCode).toBe(4403) + }) + + it('leaves the connection open when the option is not set', async () => { + const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(-10) })) + const { url } = await startServer({ verifyToken, requireAuth: true }) + + const result = await watchTicks(connect(url, bearer), 200) + + expect(result.closeCode).toBeUndefined() + expect(result.ticks).toBeGreaterThan(0) + }) + + it('leaves the connection open when the claims have no exp', async () => { + const verifyToken = vi.fn(async () => claims) + const { url } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + + const result = await watchTicks(connect(url, bearer), 200) + + expect(result.closeCode).toBeUndefined() + expect(result.ticks).toBeGreaterThan(0) + }) + + it('stops the timer when the client disconnects first', async () => { + const exp = epochSeconds(1) + const verifyToken = vi.fn(async () => ({ ...claims, exp })) + const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + const client = connect(url, bearer) + + await watchTicks(client, 50) + await client.dispose() + await delay(exp * 1000 - Date.now() + 100) + + expect(logger.info).toHaveBeenCalledWith('Subscription connection disconnected', expect.anything()) + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection auth token expired', expect.anything()) + }) + }) + + describe('keepAlive', () => { + it('pings the client at the given interval', async () => { + const { url } = await startServer({ keepAlive: 50 }) + + expect(await pingsWithin(url, 280)).toBeGreaterThanOrEqual(3) + }) + + it('sends no pings when it is 0', async () => { + const { url } = await startServer({ keepAlive: 0 }) + + expect(await pingsWithin(url, 280)).toBe(0) + }) + }) }) diff --git a/src/subscriptions/server.ts b/src/subscriptions/server.ts index 30d6a57..7b1d40c 100644 --- a/src/subscriptions/server.ts +++ b/src/subscriptions/server.ts @@ -4,12 +4,15 @@ import type { GraphQLSchema } from 'graphql' import { CloseCode } from 'graphql-ws' import { useServer } from 'graphql-ws/use/ws' import type { Server } from 'http' -import { WebSocketServer } from 'ws' +import { WebSocketServer, type WebSocket } from 'ws' import type { GraphQLContext, JwtPayload } from '../context' import { logSubscriptionOperation } from '../logging' import type { CreateSubscriptionContext, ExtractSubscriptionToken } from './context' import { defaultExtractSubscriptionToken, getHost } from './utils' +// The longest delay setTimeout accepts. A token that expires later than this is timed in steps. +const maxTimeoutMs = 2 ** 31 - 1 + export function useSubscriptionsServer({ schema, httpServer, @@ -23,6 +26,8 @@ export function useSubscriptionsServer({ jwtClaimsToLog = ['oid', 'iss'], resolveSubscriptionOperationLogger, includeDeprecatedElements, + closeOnTokenExpiry, + keepAlive = 12_000, }: { schema: GraphQLSchema httpServer: Server @@ -45,6 +50,24 @@ export function useSubscriptionsServer({ * established. Not collected per emitted payload: the usage is a property of the operation. */ includeDeprecatedElements?: boolean + /** + * If true, a connection is closed with `4403 Forbidden` when its token expires, which ends every + * subscription running on it. graphql-ws clients retry that close code, so a client reconnects with + * a fresh token. The expiry is the `exp` claim that `verifyToken` returned when the connection was + * established: the token cannot change for the life of a connection, whether it arrives in the + * connection params or on the upgrade request. A connection without a token, or whose claims have + * no `exp`, is not timed. + * + * Without it, an expired token is refused only when the client next subscribes on the connection, + * and subscriptions already running continue. + */ + closeOnTokenExpiry?: boolean + /** + * Milliseconds between the server's websocket pings. A connection that has not answered a ping by + * the time the next one is due is terminated. `0` turns the pings off. Defaults to graphql-ws's 12 + * seconds. + */ + keepAlive?: number }) { if (requireAuth && !verifyToken) throw new Error('verifyToken must be supplied when requireAuth is true') @@ -53,6 +76,28 @@ export function useSubscriptionsServer({ path, }) + const expiryTimers = new WeakMap() + const closeAtExpiry = (socket: WebSocket, claims: JwtPayload) => { + // A client that went away while its token was being verified has already passed onClose. + if (typeof claims.exp !== 'number' || socket.readyState !== socket.OPEN) return + const expiresAt = claims.exp * 1000 + const schedule = () => { + const remaining = expiresAt - Date.now() + const timer = setTimeout( + () => { + if (remaining > maxTimeoutMs) return schedule() + expiryTimers.delete(socket) + logger.info('Subscription connection auth token expired', { claims: pick(claims, jwtClaimsToLog) }) + socket.close(CloseCode.Forbidden, 'Forbidden') + }, + Math.min(Math.max(remaining, 0), maxTimeoutMs), + ) + timer.unref() + expiryTimers.set(socket, timer) + } + schedule() + } + return useServer( { schema, @@ -82,6 +127,7 @@ export function useSubscriptionsServer({ logger.info(connectionEstablished, { claims: pick(claims, jwtClaimsToLog), }) + if (closeOnTokenExpiry) closeAtExpiry(ctx.extra.socket, claims) return true } catch (error) { logger.error('Failed to verify subscription connection auth token', { error }) @@ -108,6 +154,10 @@ export function useSubscriptionsServer({ onDisconnect({ extra: { claims } }) { logger.info('Subscription connection disconnected', { claims: pick(claims, jwtClaimsToLog) }) }, + onClose({ extra: { socket } }) { + clearTimeout(expiryTimers.get(socket)) + expiryTimers.delete(socket) + }, context: async (ctx) => { return createSubscriptionContext({ connectRequest: ctx.extra.request, @@ -129,5 +179,6 @@ export function useSubscriptionsServer({ }, }, wsServer, + keepAlive, ) } From ff4564ee1e22bf9971e7aba288209a89cace25ef Mon Sep 17 00:00:00 2001 From: Sam Curry Date: Thu, 15 Oct 2026 17:54:10 +0800 Subject: [PATCH 2/5] fix(subscriptions): clear the expiry timer on the socket's close event graphql-ws's onClose waits for every subscription on the connection to return, so a subscription waiting for its next event could leave the timer running past a disconnect. The docs now say a retry needs a fresh token, and the expiry tests no longer round exp down to a whole second. --- README.md | 2 +- src/subscriptions/server.spec.ts | 27 ++++++++++++++++++++++++++- src/subscriptions/server.ts | 31 +++++++++++++++++-------------- 3 files changed, 44 insertions(+), 16 deletions(-) diff --git a/README.md b/README.md index ff07430..7c1e694 100644 --- a/README.md +++ b/README.md @@ -382,7 +382,7 @@ This library includes a `subscriptions` module to provide simple setup using the Pass `includeDeprecatedElements: true` to add [deprecation usage](#collectdeprecatedelementusage) to the operation log entry as `deprecatedElements`. It is collected once, when the subscription is established, rather than per emitted payload — the usage is a property of the operation, not of each event. - Pass `closeOnTokenExpiry: true` to close a connection with `4403 Forbidden` when the `exp` of the token it connected with passes, which ends every subscription running on it. Without it, an expired token is refused only when the client next subscribes on the connection, and subscriptions already running continue. graphql-ws clients retry `4403`, so a client reconnects, and `connectionParams` (or whatever authorises the upgrade request) supplies a fresh token. + Pass `closeOnTokenExpiry: true` to close a connection with `4403 Forbidden` when the `exp` of the token it connected with passes, which ends every subscription running on it. Without it, an expired token is refused only when the client next subscribes on the connection, and subscriptions already running continue. graphql-ws clients retry `4403` by default, but a retry succeeds only with a fresh token: pass `connectionParams` to `createClient` as a function that returns a current token, or authenticate the upgrade request afresh (by a cookie and a proxy, for example). A client that sends the same expired token fails again until its `retryAttempts` run out, and its subscriptions then end with an error. `keepAlive` sets the milliseconds between the server's websocket pings (default 12 seconds, graphql-ws's default; `0` turns them off). A connection that has not answered a ping by the time the next one is due is terminated. The pings also stop proxies and load balancers from closing an idle connection, so keep the interval below their idle timeouts. diff --git a/src/subscriptions/server.spec.ts b/src/subscriptions/server.spec.ts index 3f193b9..c2ad28e 100644 --- a/src/subscriptions/server.spec.ts +++ b/src/subscriptions/server.spec.ts @@ -30,6 +30,15 @@ const schema = new GraphQLSchema({ } }, }, + // Yields once, then waits for an event that never comes, as a subscription does between events. graphql-ws's + // onClose waits for every subscription on the connection to return, which this one cannot do while it waits. + stalls: { + type: GraphQLInt, + subscribe: async function* () { + yield { stalls: 1 } + await new Promise(() => {}) + }, + }, }, }), }) @@ -112,7 +121,8 @@ const watchTicks = (client: Client, ms: number) => }) }) -const epochSeconds = (offsetSeconds: number) => Math.floor(Date.now() / 1000) + offsetSeconds +// Not rounded to whole seconds, so a future expiry is always the full offset away. +const epochSeconds = (offsetSeconds: number) => Date.now() / 1000 + offsetSeconds const pingsWithin = async (url: string, ms: number) => { const socket = new WebSocket(url, 'graphql-transport-ws') @@ -222,6 +232,21 @@ describe('useSubscriptionsServer', () => { expect(logger.info).toHaveBeenCalledWith('Subscription connection disconnected', expect.anything()) expect(logger.info).not.toHaveBeenCalledWith('Subscription connection auth token expired', expect.anything()) }) + + it('stops the timer when the client disconnects while a subscription waits for its next event', async () => { + const exp = epochSeconds(1) + const verifyToken = vi.fn(async () => ({ ...claims, exp })) + const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + const client = connect(url, bearer) + + await new Promise((resolve) => + client.subscribe({ query: 'subscription { stalls }' }, { next: resolve, error: resolve, complete: () => {} }), + ) + await client.dispose() + await delay(exp * 1000 - Date.now() + 100) + + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection auth token expired', expect.anything()) + }) }) describe('keepAlive', () => { diff --git a/src/subscriptions/server.ts b/src/subscriptions/server.ts index 7b1d40c..68708ba 100644 --- a/src/subscriptions/server.ts +++ b/src/subscriptions/server.ts @@ -52,11 +52,16 @@ export function useSubscriptionsServer({ includeDeprecatedElements?: boolean /** * If true, a connection is closed with `4403 Forbidden` when its token expires, which ends every - * subscription running on it. graphql-ws clients retry that close code, so a client reconnects with - * a fresh token. The expiry is the `exp` claim that `verifyToken` returned when the connection was - * established: the token cannot change for the life of a connection, whether it arrives in the - * connection params or on the upgrade request. A connection without a token, or whose claims have - * no `exp`, is not timed. + * subscription running on it. The expiry is the `exp` claim that `verifyToken` returned when the + * connection was established: the token cannot change for the life of a connection, whether it + * arrives in the connection params or on the upgrade request. A connection without a token, or whose + * claims have no `exp`, is not timed. + * + * graphql-ws clients retry `4403` by default, but a retry succeeds only with a fresh token: the + * client's `connectionParams` must be a function that returns a current token, or the upgrade + * request must be authenticated afresh (by a cookie and a proxy, for example). A client that sends + * the same expired token fails again until its `retryAttempts` run out, and its subscriptions then + * end with an error. * * Without it, an expired token is refused only when the client next subscribes on the connection, * and subscriptions already running continue. @@ -76,24 +81,26 @@ export function useSubscriptionsServer({ path, }) - const expiryTimers = new WeakMap() const closeAtExpiry = (socket: WebSocket, claims: JwtPayload) => { - // A client that went away while its token was being verified has already passed onClose. + // A client that went away while its token was being verified has already closed the socket. if (typeof claims.exp !== 'number' || socket.readyState !== socket.OPEN) return const expiresAt = claims.exp * 1000 + let timer: NodeJS.Timeout | undefined + // The socket's own close event, not graphql-ws's onClose, which waits for every subscription on the connection to + // return. A subscription waiting for its next event can hold that up indefinitely. + socket.once('close', () => clearTimeout(timer)) const schedule = () => { const remaining = expiresAt - Date.now() - const timer = setTimeout( + timer = setTimeout( () => { if (remaining > maxTimeoutMs) return schedule() - expiryTimers.delete(socket) + if (socket.readyState !== socket.OPEN) return logger.info('Subscription connection auth token expired', { claims: pick(claims, jwtClaimsToLog) }) socket.close(CloseCode.Forbidden, 'Forbidden') }, Math.min(Math.max(remaining, 0), maxTimeoutMs), ) timer.unref() - expiryTimers.set(socket, timer) } schedule() } @@ -154,10 +161,6 @@ export function useSubscriptionsServer({ onDisconnect({ extra: { claims } }) { logger.info('Subscription connection disconnected', { claims: pick(claims, jwtClaimsToLog) }) }, - onClose({ extra: { socket } }) { - clearTimeout(expiryTimers.get(socket)) - expiryTimers.delete(socket) - }, context: async (ctx) => { return createSubscriptionContext({ connectRequest: ctx.extra.request, From 44d7207a5cf11623a8565fbeb873888399b08f4e Mon Sep 17 00:00:00 2001 From: Sam Curry Date: Thu, 15 Oct 2026 17:56:16 +0800 Subject: [PATCH 3/5] chore(deps): fix audit advisories in dev dependencies npm audit fix updates proxy-addr, shell-quote and source-map-js within their ranges. @graphql-tools/utils (GHSA-7mx3-vvmw-hjmv) has no fixed 8.x release, so an override moves graphql-middleware's copy to 12.0.4 in this repo's own tree. Consumers supply graphql-shield themselves and are unaffected. --- package-lock.json | 72 ++++++++++++++++++++++++++++++++++++----------- package.json | 3 +- 2 files changed, 58 insertions(+), 17 deletions(-) diff --git a/package-lock.json b/package-lock.json index f3737a7..d6f4066 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1065,15 +1065,21 @@ } }, "node_modules/@graphql-tools/utils": { - "version": "8.9.0", - "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-8.9.0.tgz", - "integrity": "sha512-pjJIWH0XOVnYGXCqej8g/u/tsfV4LvLlj0eATKQu5zwnxd/TiTHq7Cg313qUPTFFHZ3PP5wJ15chYVtLDwaymg==", + "version": "12.0.4", + "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.4.tgz", + "integrity": "sha512-hJmeBRrK1OUABqRxLXB1YwrBB+3zTXeRJYb5PaF7CweV80zNGnN640Jz+0BrCaUE8HBk6KtoVXM/tUScUsKdeg==", "dev": true, "license": "MIT", "peer": true, "dependencies": { + "@graphql-typed-document-node/core": "^3.1.1", + "@whatwg-node/promise-helpers": "^2.0.0", + "cross-inspect": "1.0.1", "tslib": "^2.4.0" }, + "engines": { + "node": ">=16.0.0" + }, "peerDependencies": { "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } @@ -2687,6 +2693,20 @@ "url": "https://opencollective.com/vitest" } }, + "node_modules/@whatwg-node/promise-helpers": { + "version": "2.0.0", + "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-2.0.0.tgz", + "integrity": "sha512-/i9CS0Z96o+cPwaYoAyVxAThDiUX1wV3tDmNWPJbn+vtdR+JftuCDg1NGCXBrzIYVDGkJ8uT6ctwmrBOUzOpKQ==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.6.3" + }, + "engines": { + "node": ">=22.15.0" + } + }, "node_modules/accepts": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/accepts/-/accepts-2.0.0.tgz", @@ -3311,6 +3331,20 @@ } } }, + "node_modules/cross-inspect": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/cross-inspect/-/cross-inspect-1.0.1.tgz", + "integrity": "sha512-Pcw1JTvZLSJH83iiGWt6fRcT+BjZlCDRVwYLbUcHzv/CRpB7r0MlSrGbIyQvVSNyGnbt7G4AXuyCiDR3POvZ1A==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.4.0" + }, + "engines": { + "node": ">=16.0.0" + } + }, "node_modules/cross-spawn": { "version": "7.0.6", "resolved": "https://registry.npmjs.org/cross-spawn/-/cross-spawn-7.0.6.tgz", @@ -6413,16 +6447,21 @@ "dev": true }, "node_modules/proxy-addr": { - "version": "2.0.7", - "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz", - "integrity": "sha512-llQsMLSUDUPT44jdrU/O37qlnifitDP+ZwrmmZcoSKyLKvtZxpyV0n2/bD/N4tBAAZ/gJEdZU7KMraoK1+XYAg==", + "version": "2.0.8", + "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.8.tgz", + "integrity": "sha512-5nnx0yGyVUcY6t9RnWcARWtwT9F1D8O9rt08htPvnd49W1IgZtmLkhu9WfMzQj1cFxjHIO6connUNVW5k7AVyQ==", "dev": true, + "license": "MIT", "dependencies": { "forwarded": "0.2.0", "ipaddr.js": "1.9.1" }, "engines": { "node": ">= 0.10" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/express" } }, "node_modules/punycode": { @@ -6898,9 +6937,9 @@ } }, "node_modules/shell-quote": { - "version": "1.10.0", - "resolved": "https://registry.npmjs.org/shell-quote/-/shell-quote-1.10.0.tgz", - "integrity": "sha512-w1aiOKwKuRgtwAReIIj89puqg+I7GvX4IbLrvmhXbzQsj1+Zwi4VO3+fa6ZF91TWSjIxoEkKnMeHcLEODK5ZXA==", + "version": "1.12.0", + "resolved": "https://registry.npmjs.org/shell-quote/-/shell-quote-1.12.0.tgz", + "integrity": "sha512-PcByqNyT/38F2kDNi006HAMRJaULuBzq/FOsw3qdZvX/GA9W/jamDaRskgHjubHiftXK5sIFxLNkvrXUwcof6Q==", "dev": true, "license": "MIT", "engines": { @@ -7024,9 +7063,9 @@ } }, "node_modules/source-map-js": { - "version": "1.2.1", - "resolved": "https://registry.npmjs.org/source-map-js/-/source-map-js-1.2.1.tgz", - "integrity": "sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==", + "version": "1.2.2", + "resolved": "https://registry.npmjs.org/source-map-js/-/source-map-js-1.2.2.tgz", + "integrity": "sha512-KGj/8Y43x35aZVDtt+J4mK1hoLGHULMYfSkODJNQjNDC3oW1PqPoxMwo0pLUsWM/UEGzON/NxeHywEfNXNP3Vw==", "dev": true, "license": "BSD-3-Clause", "engines": { @@ -7444,10 +7483,11 @@ } }, "node_modules/tslib": { - "version": "2.6.2", - "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.6.2.tgz", - "integrity": "sha512-AEYxH93jGFPn/a2iVAwW87VuUIkR1FVUKB77NwMF7nBTDkDrrT/Hpt/IrCJ0QXhW27jTBDcf5ZY7w6RiqTMw2Q==", - "dev": true + "version": "2.8.1", + "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", + "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", + "dev": true, + "license": "0BSD" }, "node_modules/tsx": { "version": "4.21.0", diff --git a/package.json b/package.json index bc63c70..f92a22b 100644 --- a/package.json +++ b/package.json @@ -96,6 +96,7 @@ "ws": "^8.20.0" }, "overrides": { - "esbuild": "^0.28.1" + "esbuild": "^0.28.1", + "@graphql-tools/utils": "^12.0.4" } } From 8d752867896c0dc75fa49046d50f263ee8be5e10 Mon Sep 17 00:00:00 2001 From: Sam Curry Date: Thu, 15 Oct 2026 18:17:59 +0800 Subject: [PATCH 4/5] feat(subscriptions): replace closeOnTokenExpiry with connectionExpiry A short-lived token can prove a longer session, such as an assertion a proxy adds to the upgrade request. Closing at the token's exp would cut subscriptions off every few minutes, so the consumer now chooses when a connection ends. Until then the token is not verified again on subscribe. An operation that fails the subscribe-time check is no longer run: onSubscribe returned nothing after closing the socket, so graphql-ws still executed it. --- README.md | 16 ++++- src/subscriptions/server.spec.ts | 100 +++++++++++++++++++++---------- src/subscriptions/server.ts | 70 +++++++++++++++------- 3 files changed, 133 insertions(+), 53 deletions(-) diff --git a/README.md b/README.md index 7c1e694..a105de0 100644 --- a/README.md +++ b/README.md @@ -382,7 +382,21 @@ This library includes a `subscriptions` module to provide simple setup using the Pass `includeDeprecatedElements: true` to add [deprecation usage](#collectdeprecatedelementusage) to the operation log entry as `deprecatedElements`. It is collected once, when the subscription is established, rather than per emitted payload — the usage is a property of the operation, not of each event. - Pass `closeOnTokenExpiry: true` to close a connection with `4403 Forbidden` when the `exp` of the token it connected with passes, which ends every subscription running on it. Without it, an expired token is refused only when the client next subscribes on the connection, and subscriptions already running continue. graphql-ws clients retry `4403` by default, but a retry succeeds only with a fresh token: pass `connectionParams` to `createClient` as a function that returns a current token, or authenticate the upgrade request afresh (by a cookie and a proxy, for example). A client that sends the same expired token fails again until its `retryAttempts` run out, and its subscriptions then end with an error. + By default a connection's token is verified when the connection is established, and again each time the client subscribes on it. A token that no longer verifies then closes the connection with `4403 Forbidden`, and the operation is not run. Subscriptions already running continue until that happens. + + To give connections a set lifetime instead, pass `connectionExpiry`. It receives the claims `verifyToken` returned and returns when the connection ends, in milliseconds since the epoch (or `undefined` for the default behaviour). At that time the connection is closed with `4403`, which ends every subscription running on it. Until then the client may subscribe again without the token being verified again: a connection's token cannot change. + + ```ts + // End each connection when its token expires + connectionExpiry: (claims) => (claims.exp === undefined ? undefined : claims.exp * 1000) + + // A short-lived token (such as a proxy's assertion) that proves a longer session: connections last an hour from when the token was issued + connectionExpiry: (claims) => (claims.iat === undefined ? undefined : (claims.iat + 3600) * 1000) + ``` + + With a lifetime longer than the token's, a revoked session is noticed only when the connection ends. + + graphql-ws clients retry `4403` by default, but a retry succeeds only with a fresh token: pass `connectionParams` to `createClient` as a function that returns a current token, or authenticate the upgrade request afresh (by a cookie and a proxy, for example). A client that sends the same expired token fails again until its `retryAttempts` run out, and its subscriptions then end with an error. `keepAlive` sets the milliseconds between the server's websocket pings (default 12 seconds, graphql-ws's default; `0` turns them off). A connection that has not answered a ping by the time the next one is due is terminated. The pings also stop proxies and load balancers from closing an idle connection, so keep the interval below their idle timeouts. diff --git a/src/subscriptions/server.spec.ts b/src/subscriptions/server.spec.ts index c2ad28e..c1132d8 100644 --- a/src/subscriptions/server.spec.ts +++ b/src/subscriptions/server.spec.ts @@ -10,8 +10,12 @@ import type { GraphQLContext, JwtPayload } from '../context' import { createSubscriptionContextFactory, type ExtractSubscriptionToken } from './context' import { useSubscriptionsServer } from './server' +// How many times the bump mutation has run, so a test can tell that a refused operation did not run. +let bumps = 0 + const schema = new GraphQLSchema({ query: new GraphQLObjectType({ name: 'Query', fields: { ok: { type: GraphQLBoolean, resolve: () => true } } }), + mutation: new GraphQLObjectType({ name: 'Mutation', fields: { bump: { type: GraphQLInt, resolve: () => ++bumps } } }), subscription: new GraphQLObjectType({ name: 'Subscription', fields: { @@ -64,6 +68,8 @@ let cleanup: Array<() => Promise | void> = [] afterEach(async () => { for (const fn of cleanup.reverse()) await fn() cleanup = [] + bumps = 0 + vi.restoreAllMocks() }) const startServer = async (options: Partial[0]>) => { @@ -121,8 +127,14 @@ const watchTicks = (client: Client, ms: number) => }) }) -// Not rounded to whole seconds, so a future expiry is always the full offset away. -const epochSeconds = (offsetSeconds: number) => Date.now() / 1000 + offsetSeconds +const bump = (client: Client) => + new Promise((resolve, reject) => { + client.subscribe({ query: 'mutation { bump }' }, { next: ({ data }) => resolve(data?.bump), error: reject, complete: () => {} }) + }) + +// A verifyToken that accepts the token when the connection is established, and refuses it afterwards, as if it had expired. +const verifiesOnce = () => + vi.fn<(host: string, token: string) => Promise>().mockResolvedValueOnce(claims).mockRejectedValue(new Error('jwt expired')) const pingsWithin = async (url: string, ms: number) => { const socket = new WebSocket(url, 'graphql-transport-ws') @@ -174,78 +186,102 @@ describe('useSubscriptionsServer', () => { expect(logger.error).toHaveBeenCalledWith('No auth token was supplied with the websocket connection') }) - describe('closeOnTokenExpiry', () => { + it('refuses an operation whose token no longer verifies, without running it', async () => { + const { url } = await startServer({ verifyToken: verifiesOnce(), requireAuth: true }) + + await expect(bump(connect(url, { connectionParams: { Authorization: 'Bearer token-params' } }))).rejects.toMatchObject({ code: 4403 }) + + expect(bumps).toBe(0) + }) + + describe('connectionExpiry', () => { const bearer = { connectionParams: { Authorization: 'Bearer token-params' } } + const endsIn = (ms: number) => () => Date.now() + ms - it('closes the connection with 4403 when the token expires, ending a running subscription', async () => { - const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(1) })) - const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + it('is given the claims that verifyToken returned', async () => { + const connectionExpiry = vi.fn(() => undefined) + const { url } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry }) + + await subscribeToToken(connect(url, bearer)) + + expect(connectionExpiry).toHaveBeenCalledWith(claims) + }) + + it('closes the connection with 4403 at the time it returns, ending a running subscription', async () => { + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(1000) }) const result = await watchTicks(connect(url, bearer), 3000) expect(result.closeCode).toBe(4403) expect(result.ticks).toBeGreaterThan(0) - expect(logger.info).toHaveBeenCalledWith('Subscription connection auth token expired', { + expect(logger.info).toHaveBeenCalledWith('Subscription connection expired', { claims: { oid: 'oid-1', iss: 'https://issuer.example' }, }) }) - it('closes the connection at once when the token has already expired', async () => { - const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(-10) })) - const { url } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + it('closes the connection at once when that time has passed', async () => { + const { url } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(-1000) }) const result = await watchTicks(connect(url, bearer), 1000) expect(result.closeCode).toBe(4403) }) - it('leaves the connection open when the option is not set', async () => { - const verifyToken = vi.fn(async () => ({ ...claims, exp: epochSeconds(-10) })) - const { url } = await startServer({ verifyToken, requireAuth: true }) + it('lets the client subscribe again until then, without verifying the token again', async () => { + const verifyToken = verifiesOnce() + const { url } = await startServer({ verifyToken, requireAuth: true, connectionExpiry: endsIn(60_000) }) + const client = connect(url, bearer) - const result = await watchTicks(connect(url, bearer), 200) + expect(await subscribeToToken(client)).toBe('token-params') + expect(await bump(client)).toBe(1) + expect(verifyToken).toHaveBeenCalledTimes(1) + }) - expect(result.closeCode).toBeUndefined() - expect(result.ticks).toBeGreaterThan(0) + it('refuses an operation that arrives after that time, without running it', async () => { + const endsAt = Date.now() + 60_000 + const { url } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: () => endsAt }) + const client = connect(url, bearer) + await subscribeToToken(client) + + // Past the end, before the timer has fired. + vi.spyOn(Date, 'now').mockReturnValue(endsAt) + + await expect(bump(client)).rejects.toMatchObject({ code: 4403 }) + expect(bumps).toBe(0) }) - it('leaves the connection open when the claims have no exp', async () => { - const verifyToken = vi.fn(async () => claims) - const { url } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + it('verifies the token on each subscribe, as by default, when it returns undefined', async () => { + const verifyToken = verifiesOnce() + const { url } = await startServer({ verifyToken, requireAuth: true, connectionExpiry: () => undefined }) - const result = await watchTicks(connect(url, bearer), 200) + await expect(subscribeToToken(connect(url, bearer))).rejects.toMatchObject({ code: 4403 }) - expect(result.closeCode).toBeUndefined() - expect(result.ticks).toBeGreaterThan(0) + expect(verifyToken).toHaveBeenCalledTimes(2) }) it('stops the timer when the client disconnects first', async () => { - const exp = epochSeconds(1) - const verifyToken = vi.fn(async () => ({ ...claims, exp })) - const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(1000) }) const client = connect(url, bearer) await watchTicks(client, 50) await client.dispose() - await delay(exp * 1000 - Date.now() + 100) + await delay(1100) expect(logger.info).toHaveBeenCalledWith('Subscription connection disconnected', expect.anything()) - expect(logger.info).not.toHaveBeenCalledWith('Subscription connection auth token expired', expect.anything()) + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection expired', expect.anything()) }) it('stops the timer when the client disconnects while a subscription waits for its next event', async () => { - const exp = epochSeconds(1) - const verifyToken = vi.fn(async () => ({ ...claims, exp })) - const { url, logger } = await startServer({ verifyToken, requireAuth: true, closeOnTokenExpiry: true }) + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(1000) }) const client = connect(url, bearer) await new Promise((resolve) => client.subscribe({ query: 'subscription { stalls }' }, { next: resolve, error: resolve, complete: () => {} }), ) await client.dispose() - await delay(exp * 1000 - Date.now() + 100) + await delay(1100) - expect(logger.info).not.toHaveBeenCalledWith('Subscription connection auth token expired', expect.anything()) + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection expired', expect.anything()) }) }) diff --git a/src/subscriptions/server.ts b/src/subscriptions/server.ts index 68708ba..f754a10 100644 --- a/src/subscriptions/server.ts +++ b/src/subscriptions/server.ts @@ -1,6 +1,6 @@ import type { Logger } from '@makerx/node-common' import { pick } from 'es-toolkit/compat' -import type { GraphQLSchema } from 'graphql' +import { GraphQLError, type GraphQLSchema } from 'graphql' import { CloseCode } from 'graphql-ws' import { useServer } from 'graphql-ws/use/ws' import type { Server } from 'http' @@ -10,7 +10,7 @@ import { logSubscriptionOperation } from '../logging' import type { CreateSubscriptionContext, ExtractSubscriptionToken } from './context' import { defaultExtractSubscriptionToken, getHost } from './utils' -// The longest delay setTimeout accepts. A token that expires later than this is timed in steps. +// The longest delay setTimeout accepts. A connection that ends later than this is timed in steps. const maxTimeoutMs = 2 ** 31 - 1 export function useSubscriptionsServer({ @@ -26,7 +26,7 @@ export function useSubscriptionsServer({ jwtClaimsToLog = ['oid', 'iss'], resolveSubscriptionOperationLogger, includeDeprecatedElements, - closeOnTokenExpiry, + connectionExpiry, keepAlive = 12_000, }: { schema: GraphQLSchema @@ -51,22 +51,33 @@ export function useSubscriptionsServer({ */ includeDeprecatedElements?: boolean /** - * If true, a connection is closed with `4403 Forbidden` when its token expires, which ends every - * subscription running on it. The expiry is the `exp` claim that `verifyToken` returned when the - * connection was established: the token cannot change for the life of a connection, whether it - * arrives in the connection params or on the upgrade request. A connection without a token, or whose - * claims have no `exp`, is not timed. + * When a connection ends, from the claims `verifyToken` returned when it was established: a time in + * milliseconds since the epoch, or `undefined` for the default behaviour below. + * + * When it returns a time: + * - At that time the connection is closed with `4403 Forbidden`, which ends every subscription + * running on it. + * - Until then the client may subscribe again on the connection, and the token is not verified + * again: it was verified when the connection was established, and it cannot change for the life + * of the connection. + * + * To end a connection when its token expires: `(claims) => (claims.exp === undefined ? undefined : claims.exp * 1000)`. + * When the token is short-lived proof of a longer session, such as an assertion that a proxy adds to + * the upgrade request, a later time keeps subscriptions from being cut off every few minutes: + * `(claims) => (claims.iat === undefined ? undefined : (claims.iat + 3600) * 1000)`. A connection then + * outlives its token, so a revoked session is noticed only when the connection ends. + * + * By default a connection has no set end. Its token is verified again each time the client + * subscribes, and a token that no longer verifies closes the connection with `4403` then. + * Subscriptions already running continue until that happens. * * graphql-ws clients retry `4403` by default, but a retry succeeds only with a fresh token: the * client's `connectionParams` must be a function that returns a current token, or the upgrade * request must be authenticated afresh (by a cookie and a proxy, for example). A client that sends * the same expired token fails again until its `retryAttempts` run out, and its subscriptions then * end with an error. - * - * Without it, an expired token is refused only when the client next subscribes on the connection, - * and subscriptions already running continue. */ - closeOnTokenExpiry?: boolean + connectionExpiry?: (claims: JwtPayload) => number | undefined /** * Milliseconds between the server's websocket pings. A connection that has not answered a ping by * the time the next one is due is terminated. `0` turns the pings off. Defaults to graphql-ws's 12 @@ -81,21 +92,24 @@ export function useSubscriptionsServer({ path, }) - const closeAtExpiry = (socket: WebSocket, claims: JwtPayload) => { + // When each connection ends, for the connections that connectionExpiry gave a time. + const connectionEnds = new WeakMap() + + const endConnectionAt = (socket: WebSocket, endsAt: number, claims: JwtPayload) => { // A client that went away while its token was being verified has already closed the socket. - if (typeof claims.exp !== 'number' || socket.readyState !== socket.OPEN) return - const expiresAt = claims.exp * 1000 + if (socket.readyState !== socket.OPEN) return + connectionEnds.set(socket, endsAt) let timer: NodeJS.Timeout | undefined // The socket's own close event, not graphql-ws's onClose, which waits for every subscription on the connection to // return. A subscription waiting for its next event can hold that up indefinitely. socket.once('close', () => clearTimeout(timer)) const schedule = () => { - const remaining = expiresAt - Date.now() + const remaining = endsAt - Date.now() timer = setTimeout( () => { if (remaining > maxTimeoutMs) return schedule() if (socket.readyState !== socket.OPEN) return - logger.info('Subscription connection auth token expired', { claims: pick(claims, jwtClaimsToLog) }) + logger.info('Subscription connection expired', { claims: pick(claims, jwtClaimsToLog) }) socket.close(CloseCode.Forbidden, 'Forbidden') }, Math.min(Math.max(remaining, 0), maxTimeoutMs), @@ -105,6 +119,13 @@ export function useSubscriptionsServer({ schedule() } + // Closes the connection, and keeps graphql-ws from running the operation: it would run it after the + // close if onSubscribe returned nothing. + const refuse = (socket: WebSocket) => { + socket.close(CloseCode.Forbidden, 'Forbidden') + return [new GraphQLError('Forbidden')] + } + return useServer( { schema, @@ -134,7 +155,8 @@ export function useSubscriptionsServer({ logger.info(connectionEstablished, { claims: pick(claims, jwtClaimsToLog), }) - if (closeOnTokenExpiry) closeAtExpiry(ctx.extra.socket, claims) + const endsAt = connectionExpiry?.(claims) + if (endsAt !== undefined) endConnectionAt(ctx.extra.socket, endsAt, claims) return true } catch (error) { logger.error('Failed to verify subscription connection auth token', { error }) @@ -143,11 +165,19 @@ export function useSubscriptionsServer({ }, onSubscribe: async (ctx) => { if (!verifyToken) return + const endsAt = connectionEnds.get(ctx.extra.socket) + if (endsAt !== undefined) { + // The token stands for the life connectionExpiry gave the connection. This covers an operation that + // arrives before the timer closes the connection. + if (Date.now() < endsAt) return + logger.warn('Subscription connection has expired', { claims: pick(ctx.extra.claims, jwtClaimsToLog) }) + return refuse(ctx.extra.socket) + } const token = extractToken({ connectRequest: ctx.extra.request, connectionParams: ctx.connectionParams }) if (!token) { if (requireAuth) { logger.error('No auth token was supplied with the websocket connection') - ctx.extra.socket.close(CloseCode.Forbidden, 'Forbidden') + return refuse(ctx.extra.socket) } return } @@ -155,7 +185,7 @@ export function useSubscriptionsServer({ await verifyToken(getHost(ctx.extra.request), token) } catch (error) { logger.warn('Subscription connection auth token is no longer valid', { claims: pick(ctx.extra.claims, jwtClaimsToLog), error }) - ctx.extra.socket.close(CloseCode.Forbidden, 'Forbidden') + return refuse(ctx.extra.socket) } }, onDisconnect({ extra: { claims } }) { From e23353f1266fdf6ace1be863d7d1c98ef2199f53 Mon Sep 17 00:00:00 2001 From: Sam Curry Date: Thu, 15 Oct 2026 18:34:22 +0800 Subject: [PATCH 5/5] fix(subscriptions): refuse a connection whose expiry has passed, and log refusals once A connection whose connectionExpiry had already passed was acknowledged and then closed, which resets a graphql-ws client's retry count, so the client reconnected without end. It is now refused before the acknowledgement. onError took graphql-ws v5's arguments, so it logged the operation id as the message and the payload as the errors. It now logs the id and the errors, and skips the refusals onSubscribe already logs. A connectionExpiry that throws is logged as such, and refuses the connection. The @graphql-tools/utils override is pinned to 12.0.1: later versions need Node.js 22.15, and this package supports Node.js 20. --- README.md | 2 +- package-lock.json | 148 +++++++++++++++++++++++++++---- package.json | 2 +- src/subscriptions/server.spec.ts | 46 ++++++++-- src/subscriptions/server.ts | 53 ++++++++--- 5 files changed, 210 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index a105de0..c8b08c9 100644 --- a/README.md +++ b/README.md @@ -384,7 +384,7 @@ This library includes a `subscriptions` module to provide simple setup using the By default a connection's token is verified when the connection is established, and again each time the client subscribes on it. A token that no longer verifies then closes the connection with `4403 Forbidden`, and the operation is not run. Subscriptions already running continue until that happens. - To give connections a set lifetime instead, pass `connectionExpiry`. It receives the claims `verifyToken` returned and returns when the connection ends, in milliseconds since the epoch (or `undefined` for the default behaviour). At that time the connection is closed with `4403`, which ends every subscription running on it. Until then the client may subscribe again without the token being verified again: a connection's token cannot change. + To give connections a set lifetime instead, pass `connectionExpiry`. It receives the claims `verifyToken` returned and returns the time the connection ends, in milliseconds since the epoch (or `undefined` for the default behaviour). At that time the connection is closed with `4403`, which ends every subscription running on it, and a time already past refuses the connection. Until then the client may subscribe again without the token being verified again: a connection's token cannot change. ```ts // End each connection when its token expires diff --git a/package-lock.json b/package-lock.json index d6f4066..590560d 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1005,6 +1005,40 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, + "node_modules/@graphql-tools/batch-execute/node_modules/@graphql-tools/utils": { + "version": "12.0.1", + "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.1.tgz", + "integrity": "sha512-8YC6jn4xYDS6YTY6xDAgQAs/nGgvzRaIGHS8GeSfVItQzNK7Ms24CQtE3EJY+amvR+tBnhRSX0N83aGj+V/RIg==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "@graphql-typed-document-node/core": "^3.1.1", + "@whatwg-node/promise-helpers": "^1.0.0", + "cross-inspect": "1.0.1", + "tslib": "^2.4.0" + }, + "engines": { + "node": ">=16.0.0" + }, + "peerDependencies": { + "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" + } + }, + "node_modules/@graphql-tools/batch-execute/node_modules/@whatwg-node/promise-helpers": { + "version": "1.3.2", + "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-1.3.2.tgz", + "integrity": "sha512-Nst5JdK47VIl9UcGwtv2Rcgyn5lWtZ0/mhRQ4G8NN2isxpq2TO30iqHzmwoJycjWuyUfg3GFXqP/gFHXeV57IA==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.6.3" + }, + "engines": { + "node": ">=16.0.0" + } + }, "node_modules/@graphql-tools/delegate": { "version": "8.8.1", "resolved": "https://registry.npmjs.org/@graphql-tools/delegate/-/delegate-8.8.1.tgz", @@ -1024,6 +1058,48 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, + "node_modules/@graphql-tools/delegate/node_modules/@graphql-tools/utils": { + "version": "12.0.1", + "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.1.tgz", + "integrity": "sha512-8YC6jn4xYDS6YTY6xDAgQAs/nGgvzRaIGHS8GeSfVItQzNK7Ms24CQtE3EJY+amvR+tBnhRSX0N83aGj+V/RIg==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "@graphql-typed-document-node/core": "^3.1.1", + "@whatwg-node/promise-helpers": "^1.0.0", + "cross-inspect": "1.0.1", + "tslib": "^2.4.0" + }, + "engines": { + "node": ">=16.0.0" + }, + "peerDependencies": { + "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" + } + }, + "node_modules/@graphql-tools/delegate/node_modules/@whatwg-node/promise-helpers": { + "version": "1.3.2", + "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-1.3.2.tgz", + "integrity": "sha512-Nst5JdK47VIl9UcGwtv2Rcgyn5lWtZ0/mhRQ4G8NN2isxpq2TO30iqHzmwoJycjWuyUfg3GFXqP/gFHXeV57IA==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.6.3" + }, + "engines": { + "node": ">=16.0.0" + } + }, + "node_modules/@graphql-tools/delegate/node_modules/@whatwg-node/promise-helpers/node_modules/tslib": { + "version": "2.8.1", + "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.8.1.tgz", + "integrity": "sha512-oJFu94HQb+KVduSUQL7wnpmqnfmLsOA/nAh6b6EH0wCEoK0/mPeXU6c3wKDV83MkOuHPRHtSXKKU99IBazS/2w==", + "dev": true, + "license": "0BSD", + "peer": true + }, "node_modules/@graphql-tools/delegate/node_modules/tslib": { "version": "2.4.1", "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.4.1.tgz", @@ -1047,6 +1123,40 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, + "node_modules/@graphql-tools/merge/node_modules/@graphql-tools/utils": { + "version": "12.0.1", + "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.1.tgz", + "integrity": "sha512-8YC6jn4xYDS6YTY6xDAgQAs/nGgvzRaIGHS8GeSfVItQzNK7Ms24CQtE3EJY+amvR+tBnhRSX0N83aGj+V/RIg==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "@graphql-typed-document-node/core": "^3.1.1", + "@whatwg-node/promise-helpers": "^1.0.0", + "cross-inspect": "1.0.1", + "tslib": "^2.4.0" + }, + "engines": { + "node": ">=16.0.0" + }, + "peerDependencies": { + "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" + } + }, + "node_modules/@graphql-tools/merge/node_modules/@whatwg-node/promise-helpers": { + "version": "1.3.2", + "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-1.3.2.tgz", + "integrity": "sha512-Nst5JdK47VIl9UcGwtv2Rcgyn5lWtZ0/mhRQ4G8NN2isxpq2TO30iqHzmwoJycjWuyUfg3GFXqP/gFHXeV57IA==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.6.3" + }, + "engines": { + "node": ">=16.0.0" + } + }, "node_modules/@graphql-tools/schema": { "version": "8.5.1", "resolved": "https://registry.npmjs.org/@graphql-tools/schema/-/schema-8.5.1.tgz", @@ -1064,16 +1174,16 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, - "node_modules/@graphql-tools/utils": { - "version": "12.0.4", - "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.4.tgz", - "integrity": "sha512-hJmeBRrK1OUABqRxLXB1YwrBB+3zTXeRJYb5PaF7CweV80zNGnN640Jz+0BrCaUE8HBk6KtoVXM/tUScUsKdeg==", + "node_modules/@graphql-tools/schema/node_modules/@graphql-tools/utils": { + "version": "12.0.1", + "resolved": "https://registry.npmjs.org/@graphql-tools/utils/-/utils-12.0.1.tgz", + "integrity": "sha512-8YC6jn4xYDS6YTY6xDAgQAs/nGgvzRaIGHS8GeSfVItQzNK7Ms24CQtE3EJY+amvR+tBnhRSX0N83aGj+V/RIg==", "dev": true, "license": "MIT", "peer": true, "dependencies": { "@graphql-typed-document-node/core": "^3.1.1", - "@whatwg-node/promise-helpers": "^2.0.0", + "@whatwg-node/promise-helpers": "^1.0.0", "cross-inspect": "1.0.1", "tslib": "^2.4.0" }, @@ -1084,6 +1194,20 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, + "node_modules/@graphql-tools/schema/node_modules/@whatwg-node/promise-helpers": { + "version": "1.3.2", + "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-1.3.2.tgz", + "integrity": "sha512-Nst5JdK47VIl9UcGwtv2Rcgyn5lWtZ0/mhRQ4G8NN2isxpq2TO30iqHzmwoJycjWuyUfg3GFXqP/gFHXeV57IA==", + "dev": true, + "license": "MIT", + "peer": true, + "dependencies": { + "tslib": "^2.6.3" + }, + "engines": { + "node": ">=16.0.0" + } + }, "node_modules/@graphql-typed-document-node/core": { "version": "3.2.0", "resolved": "https://registry.npmjs.org/@graphql-typed-document-node/core/-/core-3.2.0.tgz", @@ -2693,20 +2817,6 @@ "url": "https://opencollective.com/vitest" } }, - "node_modules/@whatwg-node/promise-helpers": { - "version": "2.0.0", - "resolved": "https://registry.npmjs.org/@whatwg-node/promise-helpers/-/promise-helpers-2.0.0.tgz", - "integrity": "sha512-/i9CS0Z96o+cPwaYoAyVxAThDiUX1wV3tDmNWPJbn+vtdR+JftuCDg1NGCXBrzIYVDGkJ8uT6ctwmrBOUzOpKQ==", - "dev": true, - "license": "MIT", - "peer": true, - "dependencies": { - "tslib": "^2.6.3" - }, - "engines": { - "node": ">=22.15.0" - } - }, "node_modules/accepts": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/accepts/-/accepts-2.0.0.tgz", diff --git a/package.json b/package.json index f92a22b..6510bf8 100644 --- a/package.json +++ b/package.json @@ -97,6 +97,6 @@ }, "overrides": { "esbuild": "^0.28.1", - "@graphql-tools/utils": "^12.0.4" + "@graphql-tools/utils": "12.0.1" } } diff --git a/src/subscriptions/server.spec.ts b/src/subscriptions/server.spec.ts index c1132d8..c41ffb1 100644 --- a/src/subscriptions/server.spec.ts +++ b/src/subscriptions/server.spec.ts @@ -186,12 +186,30 @@ describe('useSubscriptionsServer', () => { expect(logger.error).toHaveBeenCalledWith('No auth token was supplied with the websocket connection') }) - it('refuses an operation whose token no longer verifies, without running it', async () => { - const { url } = await startServer({ verifyToken: verifiesOnce(), requireAuth: true }) + it('refuses an operation whose token no longer verifies, without running it or logging a server error', async () => { + const { url, logger } = await startServer({ verifyToken: verifiesOnce(), requireAuth: true }) await expect(bump(connect(url, { connectionParams: { Authorization: 'Bearer token-params' } }))).rejects.toMatchObject({ code: 4403 }) expect(bumps).toBe(0) + expect(logger.warn).toHaveBeenCalledWith('Subscription connection auth token is no longer valid', expect.anything()) + expect(logger.error).not.toHaveBeenCalledWith('GraphQL subscriptions server error', expect.anything()) + }) + + it('logs an operation error with its id', async () => { + const { url, logger } = await startServer({}) + const client = connect(url) + + await expect( + new Promise((resolve, reject) => + client.subscribe({ query: 'subscription { nope }' }, { next: resolve, error: reject, complete: () => {} }), + ), + ).rejects.toBeDefined() + + expect(logger.error).toHaveBeenCalledWith('GraphQL subscriptions server error', { + id: expect.any(String), + errors: [expect.objectContaining({ message: 'Cannot query field "nope" on type "Subscription".' })], + }) }) describe('connectionExpiry', () => { @@ -219,12 +237,27 @@ describe('useSubscriptionsServer', () => { }) }) - it('closes the connection at once when that time has passed', async () => { - const { url } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(-1000) }) + it('refuses the connection, before acknowledging it, when that time has already passed', async () => { + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: endsIn(-1000) }) const result = await watchTicks(connect(url, bearer), 1000) - expect(result.closeCode).toBe(4403) + expect(result).toEqual({ ticks: 0, closeCode: 4403 }) + expect(logger.warn).toHaveBeenCalledWith('Subscription connection has already expired', expect.anything()) + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection established', expect.anything()) + }) + + it('refuses the connection when it throws', async () => { + const connectionExpiry = () => { + throw new Error('no iat') + } + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry }) + + const result = await watchTicks(connect(url, bearer), 1000) + + expect(result).toEqual({ ticks: 0, closeCode: 4403 }) + expect(logger.error).toHaveBeenCalledWith('Failed to compute subscription connection expiry', expect.anything()) + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection established', expect.anything()) }) it('lets the client subscribe again until then, without verifying the token again', async () => { @@ -239,7 +272,7 @@ describe('useSubscriptionsServer', () => { it('refuses an operation that arrives after that time, without running it', async () => { const endsAt = Date.now() + 60_000 - const { url } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: () => endsAt }) + const { url, logger } = await startServer({ verifyToken: async () => claims, requireAuth: true, connectionExpiry: () => endsAt }) const client = connect(url, bearer) await subscribeToToken(client) @@ -248,6 +281,7 @@ describe('useSubscriptionsServer', () => { await expect(bump(client)).rejects.toMatchObject({ code: 4403 }) expect(bumps).toBe(0) + expect(logger.error).not.toHaveBeenCalledWith('GraphQL subscriptions server error', expect.anything()) }) it('verifies the token on each subscribe, as by default, when it returns undefined', async () => { diff --git a/src/subscriptions/server.ts b/src/subscriptions/server.ts index f754a10..d66f3c5 100644 --- a/src/subscriptions/server.ts +++ b/src/subscriptions/server.ts @@ -13,6 +13,13 @@ import { defaultExtractSubscriptionToken, getHost } from './utils' // The longest delay setTimeout accepts. A connection that ends later than this is timed in steps. const maxTimeoutMs = 2 ** 31 - 1 +// What onSubscribe returns for an operation it refuses, so onError can tell a refusal from a real error. +class SubscriptionRefusedError extends GraphQLError { + constructor() { + super('Forbidden', { extensions: { code: 'FORBIDDEN' } }) + } +} + export function useSubscriptionsServer({ schema, httpServer, @@ -51,11 +58,12 @@ export function useSubscriptionsServer({ */ includeDeprecatedElements?: boolean /** - * When a connection ends, from the claims `verifyToken` returned when it was established: a time in - * milliseconds since the epoch, or `undefined` for the default behaviour below. + * Returns the time a connection ends, in milliseconds since the epoch, from the claims `verifyToken` + * returned when it was established. `undefined` keeps the default behaviour below. * * When it returns a time: - * - At that time the connection is closed with `4403 Forbidden`, which ends every subscription + * - A time already past refuses the connection with `4403 Forbidden`, as an invalid token does. + * - Otherwise at that time the connection is closed with `4403`, which ends every subscription * running on it. * - Until then the client may subscribe again on the connection, and the token is not verified * again: it was verified when the connection was established, and it cannot change for the life @@ -123,14 +131,16 @@ export function useSubscriptionsServer({ // close if onSubscribe returned nothing. const refuse = (socket: WebSocket) => { socket.close(CloseCode.Forbidden, 'Forbidden') - return [new GraphQLError('Forbidden')] + return [new SubscriptionRefusedError()] } return useServer( { schema, - onError(_ctx, message, errors) { - logger.error('GraphQL subscriptions server error', { message, errors }) + onError(_ctx, id, _payload, errors) { + // A refusal is already logged, and is not a server error. + if (errors.every((error) => error instanceof SubscriptionRefusedError)) return + logger.error('GraphQL subscriptions server error', { id, errors }) }, onConnect: async (ctx) => { const connectionEstablished = 'Subscription connection established' @@ -149,19 +159,34 @@ export function useSubscriptionsServer({ return true } + let claims: JwtPayload try { - const claims = await verifyToken(getHost(ctx.extra.request), token) - ctx.extra.claims = claims as unknown as undefined - logger.info(connectionEstablished, { - claims: pick(claims, jwtClaimsToLog), - }) - const endsAt = connectionExpiry?.(claims) - if (endsAt !== undefined) endConnectionAt(ctx.extra.socket, endsAt, claims) - return true + claims = await verifyToken(getHost(ctx.extra.request), token) } catch (error) { logger.error('Failed to verify subscription connection auth token', { error }) return false } + + let endsAt: number | undefined + try { + endsAt = connectionExpiry?.(claims) + } catch (error) { + logger.error('Failed to compute subscription connection expiry', { claims: pick(claims, jwtClaimsToLog), error }) + return false + } + // Refused before the connection is acknowledged: an acknowledged connection resets a graphql-ws + // client's retry count, so closing it at once would have the client reconnect without end. + if (endsAt !== undefined && !(endsAt > Date.now())) { + logger.warn('Subscription connection has already expired', { claims: pick(claims, jwtClaimsToLog) }) + return false + } + + ctx.extra.claims = claims as unknown as undefined + logger.info(connectionEstablished, { + claims: pick(claims, jwtClaimsToLog), + }) + if (endsAt !== undefined) endConnectionAt(ctx.extra.socket, endsAt, claims) + return true }, onSubscribe: async (ctx) => { if (!verifyToken) return