|
| 1 | +import { jest } from '@jest/globals'; |
| 2 | +import { once } from 'node:events'; |
| 3 | +import { createRequire } from 'node:module'; |
| 4 | +import SocketBroker from '../../adminforth/modules/socketBroker'; |
| 5 | +import ExpressServer from '../../adminforth/servers/express'; |
| 6 | +import { WebSocketClient } from '../../adminforth/servers/common'; |
| 7 | + |
| 8 | +const require = createRequire(new URL('../../adminforth/package.json', import.meta.url)); |
| 9 | +const express = require('express'); |
| 10 | +const { WebSocket } = require('ws'); |
| 11 | +const TOPIC = '/test/pending'; |
| 12 | +const USER = { pk: 'alice', dbUser: { id: 'alice' } }; |
| 13 | + |
| 14 | +function deferred<T>() { |
| 15 | + let resolve!: (value: T) => void; |
| 16 | + const promise = new Promise<T>((done) => { resolve = done; }); |
| 17 | + return { promise, resolve }; |
| 18 | +} |
| 19 | + |
| 20 | +function fixture(topicAuth: (topic: string, user: any) => Promise<boolean>) { |
| 21 | + const subscribed = jest.fn(); |
| 22 | + const admin = { config: { auth: { websocketTopicAuth: topicAuth, websocketSubscribed: subscribed } } }; |
| 23 | + const broker = new SocketBroker(admin as any); |
| 24 | + jest.spyOn(broker, 'startChecker').mockImplementation(async () => {}); |
| 25 | + let onMessage!: (message: string) => Promise<void>; |
| 26 | + let onClose!: () => void; |
| 27 | + const sent: any[] = []; |
| 28 | + const client = new WebSocketClient({ |
| 29 | + id: 'test-client', adminUser: USER, |
| 30 | + authorize: async () => ({ status: 'ok', adminUser: USER }), |
| 31 | + send: (message: string) => sent.push(JSON.parse(message)), |
| 32 | + close: () => onClose(), |
| 33 | + onMessage: (handler: typeof onMessage) => { onMessage = handler; }, |
| 34 | + onClose: (handler: typeof onClose) => { onClose = handler; }, |
| 35 | + }); |
| 36 | + broker.registerWsClient(client); |
| 37 | + return { |
| 38 | + broker, client, sent, subscribed, |
| 39 | + message: (type: string, topic = TOPIC) => onMessage(JSON.stringify({ type, topic })), |
| 40 | + close: () => onClose(), |
| 41 | + }; |
| 42 | +} |
| 43 | + |
| 44 | +describe('pending websocket subscriptions', () => { |
| 45 | + it('unsubscribe cancels an earlier authorization before a later publish', async () => { |
| 46 | + const auth = deferred<boolean>(); |
| 47 | + const f = fixture(() => auth.promise); |
| 48 | + const pending = f.message('subscribe'); |
| 49 | + await f.message('unsubscribe'); |
| 50 | + auth.resolve(true); |
| 51 | + await pending; |
| 52 | + await f.broker.publish(TOPIC, { after: 'unsubscribe' }); |
| 53 | + |
| 54 | + expect(f.broker.topics[TOPIC]).toBeUndefined(); |
| 55 | + expect(f.client.topics.size).toBe(0); |
| 56 | + expect(f.subscribed).not.toHaveBeenCalled(); |
| 57 | + expect(f.sent).toEqual([{ type: 'ready' }]); |
| 58 | + }); |
| 59 | + |
| 60 | + it('close cancels authorization and avoids filters for disconnected users', async () => { |
| 61 | + const auth = deferred<boolean>(); |
| 62 | + const f = fixture(() => auth.promise); |
| 63 | + const pending = f.message('subscribe'); |
| 64 | + f.close(); |
| 65 | + auth.resolve(true); |
| 66 | + await pending; |
| 67 | + await f.broker.checkDeadClients(); |
| 68 | + await f.broker.reauthorizeClients(); |
| 69 | + const filter = jest.fn(async () => true); |
| 70 | + await f.broker.publish(TOPIC, {}, filter); |
| 71 | + |
| 72 | + expect(f.broker.clients.size).toBe(0); |
| 73 | + expect(f.broker.topics[TOPIC]).toBeUndefined(); |
| 74 | + expect(f.client.topics.size).toBe(0); |
| 75 | + expect(f.subscribed).not.toHaveBeenCalled(); |
| 76 | + expect(filter).not.toHaveBeenCalled(); |
| 77 | + }); |
| 78 | + |
| 79 | + it('an older allowed subscribe cannot override a newer denied resubscribe', async () => { |
| 80 | + const older = deferred<boolean>(); |
| 81 | + const newer = deferred<boolean>(); |
| 82 | + let calls = 0; |
| 83 | + const f = fixture(() => (++calls === 1 ? older.promise : newer.promise)); |
| 84 | + const first = f.message('subscribe'); |
| 85 | + await f.message('unsubscribe'); |
| 86 | + const second = f.message('subscribe'); |
| 87 | + newer.resolve(false); |
| 88 | + await second; |
| 89 | + older.resolve(true); |
| 90 | + await first; |
| 91 | + |
| 92 | + expect(f.broker.topics[TOPIC]).toBeUndefined(); |
| 93 | + expect(f.subscribed).not.toHaveBeenCalled(); |
| 94 | + expect(f.sent).toEqual([{ type: 'ready' }, { type: 'error', message: 'Unauthorized' }]); |
| 95 | + }); |
| 96 | + |
| 97 | + it('an older denial cannot cancel a still-pending resubscribe', async () => { |
| 98 | + const older = deferred<boolean>(); |
| 99 | + const newer = deferred<boolean>(); |
| 100 | + let calls = 0; |
| 101 | + const f = fixture(() => (++calls === 1 ? older.promise : newer.promise)); |
| 102 | + const first = f.message('subscribe'); |
| 103 | + await f.message('unsubscribe'); |
| 104 | + const second = f.message('subscribe'); |
| 105 | + older.resolve(false); |
| 106 | + await first; |
| 107 | + expect(f.broker.topics[TOPIC]).toBeUndefined(); |
| 108 | + expect(f.sent).toEqual([{ type: 'ready' }]); |
| 109 | + newer.resolve(true); |
| 110 | + await second; |
| 111 | + await f.broker.publish(TOPIC, { value: 1 }); |
| 112 | + |
| 113 | + expect(f.broker.topics[TOPIC]).toEqual([f.client]); |
| 114 | + expect(f.subscribed).toHaveBeenCalledTimes(1); |
| 115 | + expect(f.sent).toEqual([{ type: 'ready' }, { type: 'message', topic: TOPIC, data: { value: 1 } }]); |
| 116 | + }); |
| 117 | +}); |
| 118 | + |
| 119 | +it('does not register a transport closed while handshake authorization was pending', async () => { |
| 120 | + const entered = deferred<void>(); |
| 121 | + const authorization = deferred<any>(); |
| 122 | + const authorize = jest.fn(async () => { |
| 123 | + entered.resolve(); |
| 124 | + return authorization.promise; |
| 125 | + }); |
| 126 | + const admin = { |
| 127 | + config: { baseUrl: '', auth: {} }, |
| 128 | + auth: { authorizeByCookies: authorize }, |
| 129 | + formatAdminForth: () => 'WebSocket regression fixture', |
| 130 | + websocket: undefined as any, |
| 131 | + }; |
| 132 | + const broker = new SocketBroker(admin as any); |
| 133 | + admin.websocket = broker; |
| 134 | + jest.spyOn(broker, 'startChecker').mockImplementation(async () => {}); |
| 135 | + const registration = jest.spyOn(broker, 'registerWsClient'); |
| 136 | + const server = new ExpressServer(admin as any) as any; |
| 137 | + server.expressApp = express(); |
| 138 | + server.setupWsServer(); |
| 139 | + server.server.listen(0, '127.0.0.1'); |
| 140 | + await once(server.server, 'listening'); |
| 141 | + const socket = new WebSocket(`ws://127.0.0.1:${server.server.address().port}/afws`); |
| 142 | + try { |
| 143 | + await once(socket, 'open'); |
| 144 | + await entered.promise; |
| 145 | + const closed = once(socket, 'close'); |
| 146 | + socket.close(); |
| 147 | + await closed; |
| 148 | + authorization.resolve({ status: 'ok', adminUser: USER }); |
| 149 | + // Drain the handshake continuation before checking its observable effects. |
| 150 | + await new Promise<void>((resolve) => setImmediate(resolve)); |
| 151 | + await broker.reauthorizeClients(); |
| 152 | + |
| 153 | + expect(registration).not.toHaveBeenCalled(); |
| 154 | + expect(broker.clients.size).toBe(0); |
| 155 | + expect(authorize).toHaveBeenCalledTimes(1); |
| 156 | + } finally { |
| 157 | + authorization.resolve({ status: 'noToken' }); |
| 158 | + socket.terminate(); |
| 159 | + await new Promise<void>((resolve) => server.server.close(resolve)); |
| 160 | + } |
| 161 | +}); |
0 commit comments