const { test } = require('node:test'); const assert = require('node:assert/strict'); const crypto = require('node:crypto'); const http = require('node:http'); const { io } = require('socket.io-client'); const { createBroker } = require('../index'); const API_KEY = 'broker-test-api-key'; const TOKEN_SECRET = 'broker-test-token-secret'; const MAX_SKEW_SECONDS = 60; const sockets = new Set(); let baseUrl = null; const broker = createBroker({ NODE_ENV: 'test', WS_PORT: '0', API_PORT: '0', API_KEY, SOCKET_TOKEN_SECRET: TOKEN_SECRET, AUTH_MODE: 'required', ALLOWED_ORIGINS: '*', }); const startPromise = broker.start().then(info => { baseUrl = info.baseUrl; return info; }); function makeToken(payload) { const encodedPayload = Buffer.from(JSON.stringify(payload)).toString('base64url'); const signature = crypto .createHmac('sha256', TOKEN_SECRET) .update(encodedPayload) .digest('base64url'); return `${encodedPayload}.${signature}`; } function connectToken(subject) { return makeToken({ v: 1, sub: subject, typ: 'connect', iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 600, }); } function topicToken(subject, topic) { return makeToken({ v: 1, sub: subject, typ: 'topic', topic, iat: Math.floor(Date.now() / 1000), exp: Math.floor(Date.now() / 1000) + 600, }); } function randomUuid() { return crypto.randomUUID(); } function connect(auth = {}) { const socket = io(baseUrl, { auth, transports: ['websocket'], reconnection: false, forceNew: true, timeout: 2000, }); sockets.add(socket); return socket; } function once(socket, event, timeoutMs = 3000) { return new Promise((resolve, reject) => { const timer = setTimeout(() => reject(new Error(`Timed out waiting for ${event}`)), timeoutMs); socket.once(event, payload => { clearTimeout(timer); resolve(payload); }); }); } function closeAll() { for (const socket of sockets) { socket.close(); } sockets.clear(); } function request(method, requestPath, options = {}) { return new Promise((resolve, reject) => { const body = options.body ? JSON.stringify(options.body) : null; const headers = { ...(options.headers || {}) }; if (body) { headers['content-type'] = 'application/json'; headers['content-length'] = Buffer.byteLength(body); } const req = http.request(`${baseUrl}${requestPath}`, { method, headers }, response => { let raw = ''; response.setEncoding('utf8'); response.on('data', chunk => { raw += chunk; }); response.on('end', () => { let parsed = raw; try { parsed = raw ? JSON.parse(raw) : null; } catch { // Keep the raw response for assertions. } resolve({ status: response.statusCode, body: parsed, raw }); }); }); req.on('error', reject); if (body) { req.write(body); } req.end(); }); } test('socket broker', async t => { try { await startPromise; await t.test('health is public and does not expose topic names', async () => { const response = await request('GET', '/health'); assert.equal(response.status, 200); assert.equal(response.body.status, 'healthy'); assert.equal(response.body.authMode, 'required'); assert.equal(response.body.tokenAuthConfigured, true); assert.equal(response.body.topics, undefined); }); await t.test('topic introspection requires the API key', async () => { const unauthorized = await request('GET', '/topics'); assert.equal(unauthorized.status, 401); const authorized = await request('GET', '/topics', { headers: { 'x-api-key': API_KEY }, }); assert.equal(authorized.status, 200); }); await t.test('connections without a token are rejected', async () => { const socket = connect({}); const error = await once(socket, 'connect_error'); assert.match(error.message, /unauthorized/i); socket.close(); }); await t.test('tampered and expired tokens are rejected', async () => { const tampered = connect({ token: `${connectToken(randomUuid())}tampered` }); const tamperedError = await once(tampered, 'connect_error'); assert.match(tamperedError.message, /unauthorized/i); tampered.close(); const expiredToken = makeToken({ v: 1, sub: randomUuid(), typ: 'connect', iat: Math.floor(Date.now() / 1000) - 3600, exp: Math.floor(Date.now() / 1000) - MAX_SKEW_SECONDS - 60, }); const expired = connect({ token: expiredToken }); const expiredError = await once(expired, 'connect_error'); assert.match(expiredError.message, /unauthorized/i); expired.close(); }); await t.test('a user can subscribe to their own notification topic only', async () => { const userId = randomUuid(); const otherUserId = randomUuid(); const socket = connect({ token: connectToken(userId) }); await once(socket, 'connect'); socket.emit('subscribe', `user/${userId}/notifications`); const subscribed = await once(socket, 'subscribed'); assert.equal(subscribed.topic, `user/${userId}/notifications`); socket.emit('subscribe', `user/${otherUserId}/notifications`); const error = await once(socket, 'error'); assert.equal(error.code, 'FORBIDDEN_TOPIC'); socket.close(); }); await t.test('a user can subscribe to their own community topic only', async () => { const userId = randomUuid(); const otherUserId = randomUuid(); const socket = connect({ token: connectToken(userId) }); await once(socket, 'connect'); socket.emit('subscribe', `community/user/${userId}`); const subscribed = await once(socket, 'subscribed'); assert.equal(subscribed.topic, `community/user/${userId}`); socket.emit('subscribe', `community/user/${otherUserId}`); const error = await once(socket, 'error'); assert.equal(error.code, 'FORBIDDEN_TOPIC'); socket.close(); }); await t.test('invalid community topics are rejected', async () => { const userId = randomUuid(); const socket = connect({ token: connectToken(userId) }); await once(socket, 'connect'); socket.emit('subscribe', 'community/user/not-a-uuid'); const error = await once(socket, 'error'); assert.equal(error.code, 'INVALID_TOPIC'); socket.close(); }); await t.test('event stats require a matching topic token', async () => { const userId = randomUuid(); const eventTopic = 'event/test-event/ticket_statistics'; const socket = connect({ token: connectToken(userId) }); await once(socket, 'connect'); socket.emit('subscribe', { topic: eventTopic, token: topicToken(userId, eventTopic), }); const subscribed = await once(socket, 'subscribed'); assert.equal(subscribed.topic, eventTopic); const otherTopic = 'event/other-event/ticket_statistics'; socket.emit('subscribe', { topic: otherTopic, token: topicToken(userId, eventTopic), }); const error = await once(socket, 'error'); assert.equal(error.code, 'FORBIDDEN_TOPIC'); socket.close(); }); await t.test('publish delivers to subscribers and unsubscribe removes the room', async () => { const userId = randomUuid(); const topic = `user/${userId}/notifications`; const socket = connect({ token: connectToken(userId) }); await once(socket, 'connect'); socket.emit('subscribe', topic); await once(socket, 'subscribed'); const messagePromise = once(socket, 'message'); const publishResponse = await request('POST', '/publish', { headers: { 'x-api-key': API_KEY }, body: { topic, message: '{"id":"notification-1"}' }, }); assert.equal(publishResponse.status, 200); const message = await messagePromise; assert.equal(message.topic, topic); assert.equal(message.message, '{"id":"notification-1"}'); socket.emit('unsubscribe', topic); await once(socket, 'unsubscribed'); const emptyResponse = await request('POST', '/publish', { headers: { 'x-api-key': API_KEY }, body: { topic, message: '{"id":"notification-2"}' }, }); assert.equal(emptyResponse.status, 204); socket.close(); }); } finally { closeAll(); await broker.stop(); } });