openvibers/openvibe.chat Commits
Cap the pre-admission WS message buffer at 200 and close overflow with…
Changes
@@@ -53,6 +53,7 @@ const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userIconst WS_HEARTBEAT_MS = 30000;const MAX_PARTICIPANTS = 8;const MAX_SOCKETS_PER_IP = 3; // one household in one channel; never one host filling it++const MAX_EARLY_MESSAGES = 200; // pre-admission buffer cap; past it the socket is closed (same as the 200/s rate limit)const FIRST_TOKEN_WAIT_MS = 1000; // how long a refused anonymous socket waits for its auth-updateconst KICK_COOLDOWN_MS = 60 * 1000;const PUBLIC_CHANNEL_ID = 'public';@@@ -319,7 +320,7 @@ class CallServer {// meanwhile wait, in order, and are handled once the socket is in the room.const early = [];let closedEarly = false;-- const onEarlyMessage = (data) => early.push(data);++ const onEarlyMessage = (data) => { if (early.length >= MAX_EARLY_MESSAGES) { try { ws.close(1008, 'Too many messages'); } catch {} return; } early.push(data); };const onEarlyClose = () => { closedEarly = true; };ws.on('message', onEarlyMessage);ws.on('close', onEarlyClose);@@@ -39,6 +39,7 @@ const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };// Concurrent chat sockets per address. Generous: one person has several tabs, a school or// carrier NAT puts many people behind one address; this only stops one host opening thousands.const MAX_CHAT_SOCKETS_PER_IP = 48;++const MAX_EARLY_MESSAGES = 200; // pre-admission buffer cap; past it the socket is closedconst DEBUG_DM_DELIVERY = process.env.DEBUG_DM_DELIVERY === '1';@@@ -281,7 +282,7 @@ class ChatServer {// meanwhile wait, in order, and are handled once the connection is registered.const early = [];let closedEarly = false;-- const onEarlyMessage = (data) => early.push(data);++ const onEarlyMessage = (data) => { if (early.length >= MAX_EARLY_MESSAGES) { try { ws.close(1008, 'Too many messages'); } catch {} return; } early.push(data); };const onEarlyClose = () => { closedEarly = true; };ws.on('message', onEarlyMessage);ws.on('close', onEarlyClose);@@@ -197,6 +197,22 @@ t('auth-update: sent before its welcome, the token admits the socket as that accawait bye(s, other);});++t('the pre-admission buffer is capped: 201 early messages close the socket (1008); a few are replayed', async () => {++ const warm = h.ctx.warm;++ h.ctx.warm = async (...a) => { await h.sleep(250); return warm.apply(h.ctx, a); };++ try {++ // 201 messages while the socket is being admitted: the buffer does not grow past the cap.++ const flood = await callWs({ channelId: 'public', token: ann.token, ip: '198.51.100.60', onOpen: (ws) => { for (let i = 0; i < 201; i++) ws.sendJson({ type: 'noop' }); } });++ assert.strictEqual(await new Promise((r) => flood.on('close', (code) => r(code))), 1008, 'closed with 1008, not buffered without bound');++ assert.ok(!flood.all.some((m) => m.type === 'welcome'), 'never admitted');++ // A few messages before admission: still admitted, and the buffer is replayed (the auth-update in it).++ const few = await callWs({ channelId: 'public', ip: '198.51.100.61', onOpen: (ws) => { ws.sendJson({ type: 'noop' }); ws.sendJson({ type: 'auth-update', token: cat.token }); } });++ const self = await few.next((m) => m.type === 'self-updated');++ assert.strictEqual(self.participant.username, 'cat', 'the early messages were replayed once admitted');++ await bye(few);++ } finally { h.ctx.warm = warm; }++});++t('a cookie call socket is cookie-reliant unless its first message is auth-update with the token (the J7 gate)', async () => {const auth = require('../server/auth/auth');const before = auth.wsCookieReliant();@@@ -62,6 +62,35 @@ t('anonymous socket upgrades to signed-in on join with a token', async () => {ws.close();});++t('the pre-admission buffer is capped: 201 early messages close the socket (1008); a few are replayed', async () => {++ const warm = h.ctx.warm;++ h.ctx.warm = async (...a) => { await h.sleep(250); return warm.apply(h.ctx, a); };++ try {++ // 201 messages while the socket is being admitted: the buffer does not grow past the cap.++ const flood = await h.ws({ ip: '198.51.100.90', token: alice.token, stream: streamId });++ const closed = new Promise((r) => flood.on('close', (code) => r(code)));++ for (let i = 0; i < 201; i++) flood.sendJson({ type: 'join', streamId });++ assert.strictEqual(await closed, 1008, 'closed with 1008, not buffered without bound');++ assert.ok(!flood.all.some((m) => m.type === 'auth'), 'never registered');++ // A few messages before admission: still registered, and the buffer is replayed in order.++ const few = await h.ws({ ip: '198.51.100.91', token: mod.token, stream: streamId });++ few.sendJson({ type: 'join', streamId, token: mod.token });++ few.sendJson({ type: 'chat', message: 'from the buffer' });++ const auth = await few.next((m) => m.type === 'auth');++ assert.strictEqual(auth.username, 'Moddy');++ const end = Date.now() + 2000;++ let saved = null;++ for (;;) {++ saved = (await h.http('GET', `/api/chat/${streamId}/history`)).body.messages.find((m) => m.message === 'from the buffer');++ if (saved) break;++ if (Date.now() > end) throw new Error('the buffered chat frame was not replayed');++ await h.sleep(20);++ }++ assert.strictEqual(saved.core_username, 'moddy', 'the early chat frame was replayed once admitted');++ few.close();++ } finally { h.ctx.warm = warm; }++});++let aliceWs, bobWs, globalWs, msgId;t('chat into a stream room: room delivery, global cross-feed, persistence, outbox, Live reactions', async () => {aliceWs = await h.ws({ ip: '198.51.100.20', token: alice.token, stream: streamId });
@@ -53,6 +53,7 @@ const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userI
@@ -53,6 +53,7 @@ const cosmetics = { getCosmeticProfile: (userId) => ctx.getCosmeticProfile(userI
const WS_HEARTBEAT_MS = 30000;
const WS_HEARTBEAT_MS = 30000;
const MAX_PARTICIPANTS = 8;
const MAX_PARTICIPANTS = 8;
const MAX_SOCKETS_PER_IP = 3; // one household in one channel; never one host filling it
const MAX_SOCKETS_PER_IP = 3; // one household in one channel; never one host filling it
+const MAX_EARLY_MESSAGES = 200; // pre-admission buffer cap; past it the socket is closed (same as the 200/s rate limit)
const FIRST_TOKEN_WAIT_MS = 1000; // how long a refused anonymous socket waits for its auth-update
const FIRST_TOKEN_WAIT_MS = 1000; // how long a refused anonymous socket waits for its auth-update
const KICK_COOLDOWN_MS = 60 * 1000;
const KICK_COOLDOWN_MS = 60 * 1000;
const PUBLIC_CHANNEL_ID = 'public';
const PUBLIC_CHANNEL_ID = 'public';
@@ -319,7 +320,7 @@ class CallServer {
@@ -319,7 +320,7 @@ class CallServer {
// meanwhile wait, in order, and are handled once the socket is in the room.
// meanwhile wait, in order, and are handled once the socket is in the room.
const early = [];
const early = [];
let closedEarly = false;
let closedEarly = false;
- const onEarlyMessage = (data) => early.push(data);
+ const onEarlyMessage = (data) => { if (early.length >= MAX_EARLY_MESSAGES) { try { ws.close(1008, 'Too many messages'); } catch {} return; } early.push(data); };
const onEarlyClose = () => { closedEarly = true; };
const onEarlyClose = () => { closedEarly = true; };
ws.on('message', onEarlyMessage);
ws.on('message', onEarlyMessage);
ws.on('close', onEarlyClose);
ws.on('close', onEarlyClose);
@@ -39,6 +39,7 @@ const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };
@@ -39,6 +39,7 @@ const tags = { getTagProfile: (userId) => ctx.getTagProfile(userId) };
// Concurrent chat sockets per address. Generous: one person has several tabs, a school or
// Concurrent chat sockets per address. Generous: one person has several tabs, a school or
// carrier NAT puts many people behind one address; this only stops one host opening thousands.
// carrier NAT puts many people behind one address; this only stops one host opening thousands.
const MAX_CHAT_SOCKETS_PER_IP = 48;
const MAX_CHAT_SOCKETS_PER_IP = 48;
+const MAX_EARLY_MESSAGES = 200; // pre-admission buffer cap; past it the socket is closed
const DEBUG_DM_DELIVERY = process.env.DEBUG_DM_DELIVERY === '1';
const DEBUG_DM_DELIVERY = process.env.DEBUG_DM_DELIVERY === '1';
@@ -281,7 +282,7 @@ class ChatServer {
@@ -281,7 +282,7 @@ class ChatServer {
// meanwhile wait, in order, and are handled once the connection is registered.
// meanwhile wait, in order, and are handled once the connection is registered.
const early = [];
const early = [];
let closedEarly = false;
let closedEarly = false;
- const onEarlyMessage = (data) => early.push(data);
+ const onEarlyMessage = (data) => { if (early.length >= MAX_EARLY_MESSAGES) { try { ws.close(1008, 'Too many messages'); } catch {} return; } early.push(data); };
const onEarlyClose = () => { closedEarly = true; };
const onEarlyClose = () => { closedEarly = true; };
ws.on('message', onEarlyMessage);
ws.on('message', onEarlyMessage);
ws.on('close', onEarlyClose);
ws.on('close', onEarlyClose);
@@ -197,6 +197,22 @@ t('auth-update: sent before its welcome, the token admits the socket as that acc
@@ -197,6 +197,22 @@ t('auth-update: sent before its welcome, the token admits the socket as that acc
await bye(s, other);
await bye(s, other);
});
});
+t('the pre-admission buffer is capped: 201 early messages close the socket (1008); a few are replayed', async () => {
+ const warm = h.ctx.warm;
+ h.ctx.warm = async (...a) => { await h.sleep(250); return warm.apply(h.ctx, a); };
+ try {
+ // 201 messages while the socket is being admitted: the buffer does not grow past the cap.
+ const flood = await callWs({ channelId: 'public', token: ann.token, ip: '198.51.100.60', onOpen: (ws) => { for (let i = 0; i < 201; i++) ws.sendJson({ type: 'noop' }); } });
+ assert.strictEqual(await new Promise((r) => flood.on('close', (code) => r(code))), 1008, 'closed with 1008, not buffered without bound');
+ assert.ok(!flood.all.some((m) => m.type === 'welcome'), 'never admitted');
+ // A few messages before admission: still admitted, and the buffer is replayed (the auth-update in it).
+ const few = await callWs({ channelId: 'public', ip: '198.51.100.61', onOpen: (ws) => { ws.sendJson({ type: 'noop' }); ws.sendJson({ type: 'auth-update', token: cat.token }); } });
+ const self = await few.next((m) => m.type === 'self-updated');
+ assert.strictEqual(self.participant.username, 'cat', 'the early messages were replayed once admitted');
+ await bye(few);
+ } finally { h.ctx.warm = warm; }
+});
+
t('a cookie call socket is cookie-reliant unless its first message is auth-update with the token (the J7 gate)', async () => {
t('a cookie call socket is cookie-reliant unless its first message is auth-update with the token (the J7 gate)', async () => {
const auth = require('../server/auth/auth');
const auth = require('../server/auth/auth');
const before = auth.wsCookieReliant();
const before = auth.wsCookieReliant();
@@ -62,6 +62,35 @@ t('anonymous socket upgrades to signed-in on join with a token', async () => {
@@ -62,6 +62,35 @@ t('anonymous socket upgrades to signed-in on join with a token', async () => {
ws.close();
ws.close();
});
});
+t('the pre-admission buffer is capped: 201 early messages close the socket (1008); a few are replayed', async () => {
+ const warm = h.ctx.warm;
+ h.ctx.warm = async (...a) => { await h.sleep(250); return warm.apply(h.ctx, a); };
+ try {
+ // 201 messages while the socket is being admitted: the buffer does not grow past the cap.
+ const flood = await h.ws({ ip: '198.51.100.90', token: alice.token, stream: streamId });
+ const closed = new Promise((r) => flood.on('close', (code) => r(code)));
+ for (let i = 0; i < 201; i++) flood.sendJson({ type: 'join', streamId });
+ assert.strictEqual(await closed, 1008, 'closed with 1008, not buffered without bound');
+ assert.ok(!flood.all.some((m) => m.type === 'auth'), 'never registered');
+ // A few messages before admission: still registered, and the buffer is replayed in order.
+ const few = await h.ws({ ip: '198.51.100.91', token: mod.token, stream: streamId });
+ few.sendJson({ type: 'join', streamId, token: mod.token });
+ few.sendJson({ type: 'chat', message: 'from the buffer' });
+ const auth = await few.next((m) => m.type === 'auth');
+ assert.strictEqual(auth.username, 'Moddy');
+ const end = Date.now() + 2000;
+ let saved = null;
+ for (;;) {
+ saved = (await h.http('GET', `/api/chat/${streamId}/history`)).body.messages.find((m) => m.message === 'from the buffer');
+ if (saved) break;
+ if (Date.now() > end) throw new Error('the buffered chat frame was not replayed');
+ await h.sleep(20);
+ }
+ assert.strictEqual(saved.core_username, 'moddy', 'the early chat frame was replayed once admitted');
+ few.close();
+ } finally { h.ctx.warm = warm; }
+});
+
let aliceWs, bobWs, globalWs, msgId;
let aliceWs, bobWs, globalWs, msgId;
t('chat into a stream room: room delivery, global cross-feed, persistence, outbox, Live reactions', async () => {
t('chat into a stream room: room delivery, global cross-feed, persistence, outbox, Live reactions', async () => {
aliceWs = await h.ws({ ip: '198.51.100.20', token: alice.token, stream: streamId });
aliceWs = await h.ws({ ip: '198.51.100.20', token: alice.token, stream: streamId });