From 158fa4538f353b917ea10384f3edb0d55c3e577b Mon Sep 17 00:00:00 2001 From: Matthew Harrigan Date: Wed, 30 Sep 2026 19:59:55 +0100 Subject: [PATCH 1/4] Do not count reauthentication towards maxConnectionsPerUser Every successful authentication, including a reauth message, goes through Sockets.prototype.auth(), which pushed the socket onto the per-user index without checking whether it was already there. Each reauthenticate() on an open connection therefore added another copy of the same socket, and a client that reauthenticates periodically (for example on access token refresh) eventually hit maxConnectionsPerUser on its own socket: the reauth was rejected with 503 and the connection was closed. Skip the index update when the socket is already indexed for that user. Co-Authored-By: Claude Opus 5.5 --- lib/listener.js | 4 ++++ test/auth.js | 30 ++++++++++++++++++++++++++++++ 2 files changed, 34 insertions(+) diff --git a/lib/listener.js b/lib/listener.js index b1ca711..1232dd8 100755 --- a/lib/listener.js +++ b/lib/listener.js @@ -452,6 +452,10 @@ internals.Sockets.prototype.auth = function (socket) { } const user = socket.auth.credentials.user; + if (this._byUser[user]?.includes(socket)) { + return; // Reauthenticated as the same user, already indexed + } + if (this._listener._settings.auth.maxConnectionsPerUser && this._byUser[user] && this._byUser[user].length >= this._listener._settings.auth.maxConnectionsPerUser) { diff --git a/test/auth.js b/test/auth.js index 3b62fbe..f523ff6 100755 --- a/test/auth.js +++ b/test/auth.js @@ -582,6 +582,36 @@ describe('authentication', () => { await server.stop(); }); + it('does not count reauthentication towards the connections limit', async () => { + + const server = Hapi.server(); + + server.auth.scheme('custom', internals.implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { index: true, maxConnectionsPerUser: 2 } } }); + await server.start(); + + const client = new Nes.Client(getUri(server.info)); + await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + for (let i = 0; i < 3; ++i) { + await client.reauthenticate({ headers: { authorization: 'Custom john' } }); + } + + const client2 = new Nes.Client(getUri(server.info)); + await client2.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + const client3 = new Nes.Client(getUri(server.info)); + await expect(client3.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } })).to.reject('Too many connections for the authenticated user'); + + client.disconnect(); + client2.disconnect(); + client3.disconnect(); + await server.stop(); + }); + it('protects an endpoint with prefix', async () => { const server = Hapi.server(); From 030e021719b73929fcec8f90796448abdc7b6cb2 Mon Sep 17 00:00:00 2001 From: Matthew Harrigan Date: Fri, 2 Oct 2026 10:30:47 +0100 Subject: [PATCH 2/4] Test reauthentication and user broadcast against the per-user index Use a minimal reauthentication test that fails on master, and add a test that a broadcast targeted at a user reaches a reauthenticated socket once (master delivers it once per reauthentication). Fixes #342 Co-Authored-By: Matt Harrison <916064+mtharrison@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- test/auth.js | 14 +++----------- test/listener.js | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 11 deletions(-) diff --git a/test/auth.js b/test/auth.js index f523ff6..3477737 100755 --- a/test/auth.js +++ b/test/auth.js @@ -596,19 +596,11 @@ describe('authentication', () => { const client = new Nes.Client(getUri(server.info)); await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); - for (let i = 0; i < 3; ++i) { - await client.reauthenticate({ headers: { authorization: 'Custom john' } }); - } - - const client2 = new Nes.Client(getUri(server.info)); - await client2.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); - - const client3 = new Nes.Client(getUri(server.info)); - await expect(client3.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } })).to.reject('Too many connections for the authenticated user'); + await client.reauthenticate({ headers: { authorization: 'Custom john' } }); + // a reauth is not a new connection + await expect(client.reauthenticate({ headers: { authorization: 'Custom john' } })).to.not.reject(); client.disconnect(); - client2.disconnect(); - client3.disconnect(); await server.stop(); }); diff --git a/test/listener.js b/test/listener.js index cd98ef3..7ddb23b 100755 --- a/test/listener.js +++ b/test/listener.js @@ -367,6 +367,45 @@ describe('Listener', () => { await server.stop(); }); + it('sends to a reauthenticated user socket once', async () => { + + const server = Hapi.server(); + + const implementation = (srv, options) => { + + return { + authenticate: (request, h) => { + + return h.authenticated({ credentials: { user: request.headers.authorization } }); + } + }; + }; + + server.auth.scheme('custom', implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + const password = 'some_not_random_password_that_is_also_long_enough'; + await server.register({ plugin: Nes, options: { auth: { type: 'direct', password, index: true } } }); + + await server.start(); + const client = new Nes.Client(getUri(server.info)); + await client.connect({ auth: { headers: { authorization: 'steve' } } }); + await client.reauthenticate({ headers: { authorization: 'steve' } }); + await client.reauthenticate({ headers: { authorization: 'steve' } }); + + const updates = []; + client.onUpdate = (update) => updates.push(update); + + server.broadcast('x', { user: 'steve' }); + + await Hoek.wait(50); + + expect(updates).to.equal(['x']); + client.disconnect(); + await server.stop(); + }); + it('errors on missing auth index (disabled)', async () => { const server = Hapi.server(); From 2bbd8092ff3c8c04d5525bfb9bab198543254f13 Mon Sep 17 00:00:00 2001 From: Matthew Harrigan Date: Mon, 5 Oct 2026 18:21:57 +0100 Subject: [PATCH 3/4] Re-index the socket when reauthentication changes the user When a reauthentication changes the credentials user, the socket now moves from the previous user to the new one in the Sockets index and in every indexed subscription it holds. Before, it stayed under the previous user: broadcast() and publish() targeted at the previous user still reached it, messages targeted at the new user did not, and on close remove() only looked at the new user, so the previous user's entry was never cleaned up and kept using one of their maxConnectionsPerUser slots. The maxConnectionsPerUser check now runs before the socket credentials are replaced. A reauthentication rejected because the new user is at the limit leaves the socket under its current user until it closes. Subscriptions made before a socket authenticated are now indexed under the user it later authenticates as. Fixes #342 Co-Authored-By: Claude Opus 5.5 --- lib/listener.js | 83 +++++++++++++++++++++++++--------- lib/socket.js | 7 ++- test/auth.js | 105 +++++++++++++++++++++++++++++++++++++++++++ test/listener.js | 113 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 287 insertions(+), 21 deletions(-) diff --git a/lib/listener.js b/lib/listener.js index 1232dd8..1d7230c 100755 --- a/lib/listener.js +++ b/lib/listener.js @@ -441,30 +441,35 @@ internals.Sockets.prototype.add = function (socket) { }; -internals.Sockets.prototype.auth = function (socket) { +internals.Sockets.prototype.auth = function (socket, previous, credentials) { - if (!this._listener._settings.auth.index) { - return; - } + // Called before the socket credentials are replaced - if (!socket.auth.credentials.user) { + if (!this._listener._settings.auth.index) { return; } - const user = socket.auth.credentials.user; - if (this._byUser[user]?.includes(socket)) { - return; // Reauthenticated as the same user, already indexed + const user = credentials.user; + if (user === previous) { + return; // Same user as before, index unchanged } - if (this._listener._settings.auth.maxConnectionsPerUser && + if (user && + this._listener._settings.auth.maxConnectionsPerUser && this._byUser[user] && this._byUser[user].length >= this._listener._settings.auth.maxConnectionsPerUser) { throw Boom.serverUnavailable('Too many connections for the authenticated user'); } - this._byUser[user] = this._byUser[user] ?? []; - this._byUser[user].push(socket); + if (previous) { + this._deindex(socket, previous); + } + + if (user) { + this._byUser[user] = this._byUser[user] ?? []; + this._byUser[user].push(socket); + } }; @@ -475,12 +480,17 @@ internals.Sockets.prototype.remove = function (socket) { if (socket.auth.credentials && socket.auth.credentials.user) { - const user = socket.auth.credentials.user; - if (this._byUser[user]) { - this._byUser[user] = this._byUser[user].filter((item) => item !== socket); - if (!this._byUser[user].length) { - delete this._byUser[user]; - } + this._deindex(socket, socket.auth.credentials.user); + } +}; + + +internals.Sockets.prototype._deindex = function (socket, user) { + + if (this._byUser[user]) { + this._byUser[user] = this._byUser[user].filter((item) => item !== socket); + if (!this._byUser[user].length) { + delete this._byUser[user]; } } }; @@ -586,15 +596,48 @@ internals.Subscribers.prototype._remove = async function (socket, path, params) }; +internals.Subscribers.prototype.reauth = function (socket, previous) { + + // Called after the socket credentials are replaced + + if (!this._settings.auth || + !this._settings.auth.index) { + + return; + } + + const user = socket.auth.credentials.user; + if (user === previous) { + return; + } + + const item = this._items[socket.id]; + if (previous) { + this._deindex(item, previous); + } + + if (user) { + this._byUser[user] = this._byUser[user] ?? []; + this._byUser[user].push(item); + } +}; + + internals.Subscribers.prototype._cleanup = function (socket, item) { delete this._items[socket.id]; if (socket.auth.credentials && - socket.auth.credentials.user && - this._byUser[socket.auth.credentials.user]) { + socket.auth.credentials.user) { + + this._deindex(item, socket.auth.credentials.user); + } +}; + + +internals.Subscribers.prototype._deindex = function (item, user) { - const user = socket.auth.credentials.user; + if (this._byUser[user]) { this._byUser[user] = this._byUser[user].filter((record) => record !== item); if (!this._byUser[user].length) { delete this._byUser[user]; diff --git a/lib/socket.js b/lib/socket.js index 783bf51..94b948e 100755 --- a/lib/socket.js +++ b/lib/socket.js @@ -598,6 +598,9 @@ internals.Socket.prototype._authByCookie = async function () { internals.Socket.prototype._setCredentials = function (auth) { + const previous = this.auth.credentials ? this.auth.credentials.user : undefined; + this._listener._sockets.auth(this, previous, auth.credentials); // Throws before any state changes when the user limit is reached + this.auth.isAuthenticated = true; this.auth.credentials = auth.credentials; this.auth.artifacts = auth.artifacts; @@ -605,7 +608,9 @@ internals.Socket.prototype._setCredentials = function (auth) { this.auth._verified = Date.now(); - this._listener._sockets.auth(this); + for (const subscribers of new Set(Object.values(this._subscriptions))) { // Paths on the same route share one subscribers object + subscribers.reauth(this, previous); + } }; diff --git a/test/auth.js b/test/auth.js index 3477737..d5908ea 100755 --- a/test/auth.js +++ b/test/auth.js @@ -604,6 +604,111 @@ describe('authentication', () => { await server.stop(); }); + it('moves the socket to the new user when reauthenticating as a different user', async () => { + + const server = Hapi.server(); + + server.auth.scheme('custom', internals.implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { index: true, maxConnectionsPerUser: 1 } } }); + await server.start(); + + const client = new Nes.Client(getUri(server.info)); + const updates = []; + client.onUpdate = (update) => updates.push(update); + await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + await client.reauthenticate({ headers: { authorization: 'Custom ed' } }); + + server.broadcast('for john', { user: 'john' }); + server.broadcast('for ed', { user: 'ed' }); + await Hoek.wait(50); + expect(updates).to.equal(['for ed']); + + // john's only slot is free again while the switched socket is still open + const client2 = new Nes.Client(getUri(server.info)); + await client2.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + client.disconnect(); + client2.disconnect(); + await Hoek.wait(50); + expect(server.plugins.nes._listener._sockets._byUser).to.equal({}); + await server.stop(); + }); + + it('rejects reauthentication as a different user who is at the connections limit', async () => { + + const server = Hapi.server(); + + server.auth.scheme('custom', internals.implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { index: true, maxConnectionsPerUser: 1 } } }); + await server.start(); + + const edClient = new Nes.Client(getUri(server.info)); + const edUpdates = []; + edClient.onUpdate = (update) => edUpdates.push(update); + await edClient.connect({ reconnect: false, auth: { headers: { authorization: 'Custom ed' } } }); + + const client = new Nes.Client(getUri(server.info)); + await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + const team = new Teamwork.Team(); + client.onDisconnect = () => team.attend(); + + await expect(client.reauthenticate({ headers: { authorization: 'Custom ed' } })).to.reject('Too many connections for the authenticated user'); + await team.work; + await Hoek.wait(50); + + // The rejected socket stayed indexed under john until it closed, so john's slot is free again + const client2 = new Nes.Client(getUri(server.info)); + await client2.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + server.broadcast('for ed', { user: 'ed' }); + await Hoek.wait(50); + expect(edUpdates).to.equal(['for ed']); + + edClient.disconnect(); + client2.disconnect(); + await Hoek.wait(50); + expect(server.plugins.nes._listener._sockets._byUser).to.equal({}); + await server.stop(); + }); + + it('updates the user index when reauthenticating to and from credentials without a user', async () => { + + const server = Hapi.server(); + + server.auth.scheme('custom', internals.implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { index: true } } }); + await server.start(); + + const client = new Nes.Client(getUri(server.info)); + const updates = []; + client.onUpdate = (update) => updates.push(update); + await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + + await client.reauthenticate({ headers: { authorization: 'Custom app' } }); + server.broadcast('for john', { user: 'john' }); + await Hoek.wait(50); + expect(updates).to.equal([]); + expect(server.plugins.nes._listener._sockets._byUser).to.equal({}); + + await client.reauthenticate({ headers: { authorization: 'Custom john' } }); + server.broadcast('for john again', { user: 'john' }); + await Hoek.wait(50); + expect(updates).to.equal(['for john again']); + + client.disconnect(); + await server.stop(); + }); + it('protects an endpoint with prefix', async () => { const server = Hapi.server(); diff --git a/test/listener.js b/test/listener.js index 7ddb23b..6fc4b2d 100755 --- a/test/listener.js +++ b/test/listener.js @@ -954,6 +954,119 @@ describe('Listener', () => { await server.stop(); }); + it('publishes to the new user after reauthenticating as a different user', async () => { + + const server = Hapi.server(); + + const implementation = (srv, options) => { + + return { + authenticate: (request, h) => { + + const name = request.headers.authorization; + return h.authenticated({ credentials: name === 'app' ? { app: name } : { user: name } }); + } + }; + }; + + server.auth.scheme('custom', implementation); + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { type: 'direct', index: true } } }); + server.subscription('/item/{id}', { auth: { index: true } }); + server.subscription('/plain'); + server.subscription('/unindexed', { auth: { mode: 'required' } }); + + await server.start(); + const edClient = new Nes.Client(getUri(server.info)); + await edClient.connect({ auth: { headers: { authorization: 'ed' } } }); + const edUpdates = []; + await edClient.subscribe('/item/1', (update) => edUpdates.push(update)); + + const client = new Nes.Client(getUri(server.info)); + await client.connect({ auth: { headers: { authorization: 'john' } } }); + + const updates = []; + const handler = (update) => updates.push(update); + await client.subscribe('/item/1', handler); + await client.subscribe('/item/2', handler); + await client.subscribe('/plain', handler); + await client.subscribe('/unindexed', handler); + + const subscribers = Object.values(server.plugins.nes._listener._sockets._items)[0]._subscriptions['/item/1']; + + await client.reauthenticate({ headers: { authorization: 'john' } }); + expect(subscribers._byUser.john).to.have.length(1); + + await client.reauthenticate({ headers: { authorization: 'ed' } }); + + await server.publish('/item/1', 'for john', { user: 'john' }); + await server.publish('/item/1', 'for ed', { user: 'ed' }); + await server.publish('/plain', 'plain'); + await server.publish('/unindexed', 'unindexed'); + await Hoek.wait(50); + expect(updates).to.equal(['for ed', 'plain', 'unindexed']); + expect(edUpdates).to.equal(['for ed']); + + await client.reauthenticate({ headers: { authorization: 'app' } }); + await server.publish('/item/2', 'for ed after app', { user: 'ed' }); + await Hoek.wait(50); + expect(subscribers._byUser.ed).to.have.length(1); + + await client.reauthenticate({ headers: { authorization: 'john' } }); + await server.publish('/item/2', 'for john again', { user: 'john' }); + await Hoek.wait(50); + expect(updates).to.equal(['for ed', 'plain', 'unindexed', 'for john again']); + + client.disconnect(); + edClient.disconnect(); + await Hoek.wait(50); + expect(subscribers._byUser).to.equal({}); + await server.stop(); + }); + + it('publishes to a subscriber that authenticated after subscribing', async () => { + + const server = Hapi.server(); + + const implementation = (srv, options) => { + + return { + authenticate: (request, h) => { + + return h.authenticated({ credentials: { user: request.headers.authorization } }); + } + }; + }; + + server.auth.scheme('custom', implementation); + server.auth.strategy('default', 'custom'); + server.auth.default({ strategy: 'default', mode: 'optional' }); + + await server.register({ plugin: Nes, options: { auth: { type: 'direct', index: true } } }); + server.subscription('/item/{id}', { auth: { mode: 'optional', index: true } }); + + await server.start(); + const client = new Nes.Client(getUri(server.info)); + await client.connect(); + + const updates = []; + await client.subscribe('/item/1', (update) => updates.push(update)); + const subscribers = Object.values(server.plugins.nes._listener._sockets._items)[0]._subscriptions['/item/1']; + expect(subscribers._byUser).to.equal({}); + + await client.reauthenticate({ headers: { authorization: 'john' } }); + await server.publish('/item/1', 'for john', { user: 'john' }); + await Hoek.wait(50); + expect(updates).to.equal(['for john']); + + client.disconnect(); + await Hoek.wait(50); + expect(subscribers._byUser).to.equal({}); + await server.stop(); + }); + it('publishes to selected user (ignores non-user credentials)', async () => { const server = Hapi.server(); From f254c4ef0a3ea9d74bd03075fc7b5f45ee1c0132 Mon Sep 17 00:00:00 2001 From: Nicolas Morel Date: Wed, 7 Oct 2026 16:20:56 +0200 Subject: [PATCH 4/4] fix: prevent race condition between reauth and close --- lib/listener.js | 8 ++++++- test/auth.js | 63 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/lib/listener.js b/lib/listener.js index 1d7230c..f00aaca 100755 --- a/lib/listener.js +++ b/lib/listener.js @@ -445,7 +445,9 @@ internals.Sockets.prototype.auth = function (socket, previous, credentials) { // Called before the socket credentials are replaced - if (!this._listener._settings.auth.index) { + if (!this._listener._settings.auth.index || + !this._items[socket.id]) { // Closed while authenticating + return; } @@ -612,6 +614,10 @@ internals.Subscribers.prototype.reauth = function (socket, previous) { } const item = this._items[socket.id]; + if (!item) { + return; // Removed while the socket was closing + } + if (previous) { this._deindex(item, previous); } diff --git a/test/auth.js b/test/auth.js index d5908ea..301d750 100755 --- a/test/auth.js +++ b/test/auth.js @@ -678,6 +678,69 @@ describe('authentication', () => { await server.stop(); }); + it('does not index a socket whose reauthentication completes after it closed', async () => { + + const server = Hapi.server(); + + const authEntered = new Teamwork.Team(); + const authGate = new Teamwork.Team(); + const unsubEntered = new Teamwork.Team(); + const unsubGate = new Teamwork.Team(); + + server.auth.scheme('custom', () => ({ + authenticate: async (request, h) => { + + const user = request.headers.authorization.split(' ')[1]; + if (user === 'ed') { + authEntered.attend(); + await authGate.work; + } + + return h.authenticated({ credentials: { user } }); + } + })); + + server.auth.strategy('default', 'custom'); + server.auth.default('default'); + + await server.register({ plugin: Nes, options: { auth: { index: true } } }); + server.subscription('/item', { + auth: { index: true }, + onUnsubscribe: async () => { + + unsubEntered.attend(); + await unsubGate.work; + } + }); + + await server.start(); + + const client = new Nes.Client(getUri(server.info)); + await client.connect({ reconnect: false, auth: { headers: { authorization: 'Custom john' } } }); + await client.subscribe('/item', Hoek.ignore); + + const listener = server.plugins.nes._listener; + const subscribers = Object.values(listener._sockets._items)[0]._subscriptions['/item']; + + // Reauth is stuck in the auth scheme while the connection closes + const reauth = client.reauthenticate({ headers: { authorization: 'Custom ed' } }).catch(Hoek.ignore); + await authEntered.work; + client.disconnect(); + + // The close handler already removed the socket and is now awaiting onUnsubscribe + await unsubEntered.work; + authGate.attend(); + await reauth; + await Hoek.wait(50); + + unsubGate.attend(); + await Hoek.wait(50); + + expect(listener._sockets._byUser).to.equal({}); + expect(subscribers._byUser).to.equal({}); + await server.stop(); + }); + it('updates the user index when reauthenticating to and from credentials without a user', async () => { const server = Hapi.server();