diff --git a/lib/listener.js b/lib/listener.js index b1ca711..f00aaca 100755 --- a/lib/listener.js +++ b/lib/listener.js @@ -441,26 +441,37 @@ internals.Sockets.prototype.add = function (socket) { }; -internals.Sockets.prototype.auth = function (socket) { +internals.Sockets.prototype.auth = function (socket, previous, credentials) { + + // Called before the socket credentials are replaced + + if (!this._listener._settings.auth.index || + !this._items[socket.id]) { // Closed while authenticating - if (!this._listener._settings.auth.index) { return; } - if (!socket.auth.credentials.user) { - return; + const user = credentials.user; + if (user === previous) { + return; // Same user as before, index unchanged } - const user = socket.auth.credentials.user; - 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); + } }; @@ -471,12 +482,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]; } } }; @@ -582,15 +598,52 @@ 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 (!item) { + return; // Removed while the socket was closing + } + + 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 3b62fbe..301d750 100755 --- a/test/auth.js +++ b/test/auth.js @@ -582,6 +582,196 @@ 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' } } }); + + 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(); + 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('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(); + + 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 cd98ef3..6fc4b2d 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(); @@ -915,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();