299 lines
9.4 KiB
JavaScript
299 lines
9.4 KiB
JavaScript
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();
|
|
}
|
|
});
|