diff --git a/README.md b/README.md index 23f6c4e..c8b08c9 100644 --- a/README.md +++ b/README.md @@ -382,6 +382,24 @@ 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. + 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 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 + 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. + Example for Apollo Server (`wsServerCleanup` called in the `drainServer` plugin callback): ```ts diff --git a/package-lock.json b/package-lock.json index 150cc77..590560d 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", @@ -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,20 +1174,40 @@ "graphql": "^14.0.0 || ^15.0.0 || ^16.0.0 || ^17.0.0" } }, - "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==", + "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": "^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/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", @@ -3311,6 +3441,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 +6557,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 +7047,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 +7173,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 +7593,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 48ba42c..6510bf8 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", @@ -96,6 +96,7 @@ "ws": "^8.20.0" }, "overrides": { - "esbuild": "^0.28.1" + "esbuild": "^0.28.1", + "@graphql-tools/utils": "12.0.1" } } diff --git a/src/subscriptions/server.spec.ts b/src/subscriptions/server.spec.ts index 2f388e2..c41ffb1 100644 --- a/src/subscriptions/server.spec.ts +++ b/src/subscriptions/server.spec.ts @@ -1,16 +1,21 @@ 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' 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: { @@ -20,6 +25,24 @@ 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) + } + }, + }, + // 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(() => {}) + }, + }, }, }), }) @@ -45,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]>) => { @@ -79,6 +104,48 @@ 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 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') + 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 +185,151 @@ describe('useSubscriptionsServer', () => { expect(verifyToken).not.toHaveBeenCalled() 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 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', () => { + const bearer = { connectionParams: { Authorization: 'Bearer token-params' } } + const endsIn = (ms: number) => () => Date.now() + ms + + 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 expired', { + claims: { oid: 'oid-1', iss: 'https://issuer.example' }, + }) + }) + + 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).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 () => { + const verifyToken = verifiesOnce() + const { url } = await startServer({ verifyToken, requireAuth: true, connectionExpiry: endsIn(60_000) }) + const client = connect(url, bearer) + + expect(await subscribeToToken(client)).toBe('token-params') + expect(await bump(client)).toBe(1) + expect(verifyToken).toHaveBeenCalledTimes(1) + }) + + it('refuses an operation that arrives after that time, without running it', async () => { + const endsAt = Date.now() + 60_000 + const { url, logger } = 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) + 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 () => { + const verifyToken = verifiesOnce() + const { url } = await startServer({ verifyToken, requireAuth: true, connectionExpiry: () => undefined }) + + await expect(subscribeToToken(connect(url, bearer))).rejects.toMatchObject({ code: 4403 }) + + expect(verifyToken).toHaveBeenCalledTimes(2) + }) + + it('stops the timer when the client disconnects first', async () => { + 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(1100) + + expect(logger.info).toHaveBeenCalledWith('Subscription connection disconnected', 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 { 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(1100) + + expect(logger.info).not.toHaveBeenCalledWith('Subscription connection 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..d66f3c5 100644 --- a/src/subscriptions/server.ts +++ b/src/subscriptions/server.ts @@ -1,15 +1,25 @@ 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' -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 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, @@ -23,6 +33,8 @@ export function useSubscriptionsServer({ jwtClaimsToLog = ['oid', 'iss'], resolveSubscriptionOperationLogger, includeDeprecatedElements, + connectionExpiry, + keepAlive = 12_000, }: { schema: GraphQLSchema httpServer: Server @@ -45,6 +57,41 @@ export function useSubscriptionsServer({ * established. Not collected per emitted payload: the usage is a property of the operation. */ includeDeprecatedElements?: boolean + /** + * 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: + * - 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 + * 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. + */ + 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 + * seconds. + */ + keepAlive?: number }) { if (requireAuth && !verifyToken) throw new Error('verifyToken must be supplied when requireAuth is true') @@ -53,11 +100,47 @@ export function useSubscriptionsServer({ path, }) + // 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 (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 = endsAt - Date.now() + timer = setTimeout( + () => { + if (remaining > maxTimeoutMs) return schedule() + if (socket.readyState !== socket.OPEN) return + logger.info('Subscription connection expired', { claims: pick(claims, jwtClaimsToLog) }) + socket.close(CloseCode.Forbidden, 'Forbidden') + }, + Math.min(Math.max(remaining, 0), maxTimeoutMs), + ) + timer.unref() + } + 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 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' @@ -76,25 +159,50 @@ 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), - }) - 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 + 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 } @@ -102,7 +210,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 } }) { @@ -129,5 +237,6 @@ export function useSubscriptionsServer({ }, }, wsServer, + keepAlive, ) }