From dd91c17fe13145bcc9deaf8fc88c63be824c33da Mon Sep 17 00:00:00 2001 From: Alec Gibson <12036746+alecgibson@users.noreply.github.com> Date: Tue, 6 Oct 2026 15:08:32 +0000 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=94=92=EF=B8=8F=20Only=20accept=20pre?= =?UTF-8?q?sence=20for=20the=20doc=20on=20its=20channel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes https://github.com/share/sharedb/issues/738 At the moment, the server checks read access against a presence's `c` and `d`, but broadcasts it on its `ch`: 1. Deny `docs/secret` in `readSnapshots` 2. Send presence on `docs.secret` with the `c`/`d` of a readable doc 3. `docs/secret` subscribers apply it This change rejects doc presence whose `c` and `d` don't match its channel, and `DocPresence` ignores presence for any other doc, which also covers untyped presence and dotted-name channel collisions. Another client's presence, or a forwarded `sendPresence` error, no longer reaches a local presence with the same ID either: the client's own replies carry its `src` if they succeed, or the request's `pv` if they fail, and forwarded errors carry neither. Subscribing, write access and presence ID ownership are left to middleware, which the new "Access control" docs cover. The server check only holds once every instance is upgraded, and the client filter only protects upgraded clients. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- docs/api/backend.md | 11 ++ docs/api/presence.md | 2 + docs/middleware/actions.md | 6 + docs/presence.md | 83 +++++++++ lib/agent.js | 12 ++ lib/client/presence/doc-presence.js | 6 + lib/client/presence/presence.js | 5 +- test/agent.js | 14 ++ test/client/presence/doc-presence.js | 240 ++++++++++++++++++++++++++- test/client/presence/presence.js | 47 ++++++ 10 files changed, 417 insertions(+), 9 deletions(-) diff --git a/docs/api/backend.md b/docs/api/backend.md index b3efa8b04..f6657783b 100644 --- a/docs/api/backend.md +++ b/docs/api/backend.md @@ -113,6 +113,17 @@ Optional {: .d-inline-block } +`doNotForwardSendPresenceErrorsToClient` -- boolean + +Optional +{: .label .label-grey } + +> Default: `false` + +> If set to `true`, errors from the [`'sendPresence'`]({{ site.baseurl }}{% link middleware/actions.md %}#sendpresence) middleware will be passed to the `errorHandler` instead of being sent to the client. Sending these errors to the client is deprecated + +{: .d-inline-block } + `doNotCommitNoOps` -- boolean Optional diff --git a/docs/api/presence.md b/docs/api/presence.md index d8768a7a9..2cc136662 100644 --- a/docs/api/presence.md +++ b/docs/api/presence.md @@ -92,6 +92,8 @@ Optional {: .warn } > Depending on use-case, the same client may have **multiple** presences, so a user or client ID may not be appropriate to use as a presence ID. +> +> Presence IDs are sent to every subscriber, so the server should check that a client only uses its own. See [Access control]({{ site.baseurl }}{% link presence.md %}#presence-ids). Return value diff --git a/docs/middleware/actions.md b/docs/middleware/actions.md index 3d34dc6a3..ef3e7cb21 100644 --- a/docs/middleware/actions.md +++ b/docs/middleware/actions.md @@ -69,6 +69,9 @@ This action has these additional `context` properties: Presence information has just been received from a client. The presence has not yet been transformed against any ops it has missed. +{: .warn } +Presence is not covered by document permissions. See [Access control]({{ site.baseurl }}{% link presence.md %}#access-control). + This action has these additional `context` properties: `collection` -- string @@ -83,6 +86,9 @@ This action has these additional `context` properties: Presence information is about to be sent to a client. +{: .warn } +Presence is not covered by document permissions. See [Access control]({{ site.baseurl }}{% link presence.md %}#access-control). + This action has these additional `context` properties: `collection` -- string diff --git a/docs/presence.md b/docs/presence.md index 9567446e2..ddf817e77 100644 --- a/docs/presence.md +++ b/docs/presence.md @@ -76,3 +76,86 @@ const localPresence = presence.create() // The presence value depends on the type localPresence.submit(value) ``` + +## Access control + +{: .warn } +Presence is **not** covered by document permissions: being able to read or write a document does not control who can see or send presence on it. Use [middleware]({{ site.baseurl }}{% link middleware/index.md %}) to restrict presence. + +### Who can see presence + +Subscribing to presence doesn't trigger any presence middleware, so any client can subscribe to any presence channel -- including the presence of a document it can't read. + +To stop a client receiving presence, reject the update in the [`'sendPresence'`]({{ site.baseurl }}{% link middleware/actions.md %}#sendpresence) middleware. For typed presence, ShareDB makes sure that `presence.c` and `presence.d` match the presence channel, so you can check read access with them. This middleware runs for every update about to be sent to every client, so keep the check cheap -- for example, by caching permissions on [`agent.custom`]({{ site.baseurl }}{% link api/agent.md %}#custom--object). + +{: .warn } +Rejecting updates in `'sendPresence'` only stops them reaching the client if you set the [`doNotForwardSendPresenceErrorsToClient`]({{ site.baseurl }}{% link api/backend.md %}#options) option. Otherwise, the client is sent an error for each rejected update, including its presence ID. With the option set, each rejected update is passed to the `errorHandler` instead, which logs it by default. + +### Who can send presence + +ShareDB rejects typed presence whose collection and document ID don't match its channel. It also only broadcasts typed presence if the client can read the document, as determined by the [`'readSnapshots'`]({{ site.baseurl }}{% link middleware/actions.md %}#readsnapshots) middleware. This is checked when the client starts sending presence on the document, not on every update. + +ShareDB doesn't check: + + - whether the client can write to the document + - `null` presence, which is sent when a client leaves + - untyped presence + +Check these in the [`'receivePresence'`]({{ site.baseurl }}{% link middleware/actions.md %}#receivepresence) middleware if you need to. + +If you only use typed presence, reject untyped presence in `'receivePresence'`: older clients still apply untyped presence sent on a document's channel. + +Older clients also apply presence for another document whose collection and ID join to the same channel, which can happen when collection names contain `.` -- for example, `docs.a`/`b` and `docs`/`a.b` both use the `docs.a.b` channel. The server can't tell these apart, so avoid `.` in the names of collections that use presence until all clients are upgraded. + +### Presence IDs + +Presence IDs are chosen by the client, and sent to every subscriber, so a client could use another client's presence ID to overwrite or clear its presence, or to stop its later updates being applied. Older clients also fire their pending submit callbacks early when they receive presence with their own presence ID. + +To prevent this, tie presence IDs to the user in `'receivePresence'` -- for example by requiring them to start with the user ID and a separator that can't appear in user IDs: + +```js +const localPresence = presence.create(`${userId}:${randomId}`) +``` + +{: .info } +Keep a random suffix, so that the same user can have more than one presence -- for example, in multiple browser tabs. + +### Example + +```js +const backend = new Backend({ + presence: true, + doNotForwardSendPresenceErrorsToClient: true, + errorHandler: (error) => { + if (error.code !== 'PRESENCE_FORBIDDEN') console.error(error) + } +}) + +backend.use('receivePresence', (context, next) => { + // agent.custom is usually set in the 'connect' hook + const userId = context.agent.custom.userId + const presence = context.presence + if (!userId || !presence.id.startsWith(`${userId}:`)) { + return next(new Error('Unauthorized')) + } + // This app only uses typed presence + if (!presence.c) { + return next(new Error('Unauthorized')) + } + // Let users clear their own presence, even if they've lost access + if (presence.p === null) return next() + if (!userCanChangeDoc(userId, presence.c, presence.d)) { + return next(new Error('Unauthorized')) + } + next() +}) + +backend.use('sendPresence', (context, next) => { + const userId = context.agent.custom.userId + const presence = context.presence + if (!userCanReadDoc(userId, presence.c, presence.d)) { + return next({code: 'PRESENCE_FORBIDDEN', message: 'Forbidden'}) + } + next() +}) +``` diff --git a/lib/agent.js b/lib/agent.js index 230928852..998b22927 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -5,6 +5,7 @@ var ACTIONS = require('./message-actions').ACTIONS; var types = require('./types'); var util = require('./util'); var protocol = require('./protocol'); +var DocPresence = require('./client/presence/doc-presence'); var ERROR_CODE = ShareDBError.CODES; @@ -466,8 +467,19 @@ Agent.prototype._checkRequest = function(request) { if (request.c != null && isInvalidName(request.c)) return 'Invalid collection'; if (request.ch != null && isInvalidName(request.ch)) return 'Invalid presence channel'; if (request.seq != null && typeof request.seq !== 'number') return 'Invalid seq'; + if (request.a === ACTIONS.presence && isDocPresence(request) && !matchesDocPresenceChannel(request)) { + return 'Presence collection and doc ID do not match channel'; + } }; +function isDocPresence(request) { + return request.c != null || request.d != null; +} + +function matchesDocPresenceChannel(request) { + return !!request.c && !!request.d && request.ch === DocPresence.channel(request.c, request.d); +} + // Handle an incoming message from the client Agent.prototype._handleMessage = function(request, callback) { try { diff --git a/lib/client/presence/doc-presence.js b/lib/client/presence/doc-presence.js index 612893c59..ced87d817 100644 --- a/lib/client/presence/doc-presence.js +++ b/lib/client/presence/doc-presence.js @@ -17,6 +17,12 @@ DocPresence.channel = function(collection, id) { return collection + '.' + id; }; +DocPresence.prototype._receiveUpdate = function(error, message) { + var isForThisDoc = message.c === this.collection && message.d === this.id; + if (!error && !isForThisDoc) return; + Presence.prototype._receiveUpdate.call(this, error, message); +}; + DocPresence.prototype._createLocalPresence = function(id) { return new LocalDocPresence(this, id); }; diff --git a/lib/client/presence/presence.js b/lib/client/presence/presence.js index 7b3fccabc..1056ddcf5 100644 --- a/lib/client/presence/presence.js +++ b/lib/client/presence/presence.js @@ -120,10 +120,13 @@ Presence.prototype._handleUnsubscribe = function(error, seq) { }; Presence.prototype._receiveUpdate = function(error, message) { + var isErrorForOwnRequest = !!error && message.pv != null; + var isFromThisConnection = message.src === this.connection.id; var localPresence = util.dig(this.localPresences, message.id); - if (localPresence) return localPresence._ack(error, message.pv); + if (localPresence && (isErrorForOwnRequest || isFromThisConnection)) return localPresence._ack(error, message.pv); if (error) return this.emit('error', error); + if (localPresence) return; var presence = this; var remotePresence = util.digOrCreate(this._remotePresenceInstances, message.id, function() { return presence._createRemotePresence(message.id); diff --git a/test/agent.js b/test/agent.js index b364b7df8..75574b636 100644 --- a/test/agent.js +++ b/test/agent.js @@ -144,6 +144,20 @@ describe('Agent', function() { }); }); + [unconvertible, ['dogs'], '__proto__'].forEach(function(badCollection) { + it('rejects doc presence with collection ' + JSON.stringify(badCollection), function(done) { + var message = {a: ACTIONS.presence, ch: 'dogs.fido', c: badCollection, d: 'fido', id: 'rex', pv: 0, v: 0}; + expectRejected(message, 'Invalid collection', done); + }); + }); + + [unconvertible, '__proto__'].forEach(function(badChannel) { + it('rejects doc presence with channel ' + JSON.stringify(badChannel), function(done) { + var message = {a: ACTIONS.presence, ch: badChannel, c: 'dogs', d: 'fido', id: 'rex', pv: 0, v: 0}; + expectRejected(message, 'Invalid presence channel', done); + }); + }); + [unconvertible, true].forEach(function(badProtocol) { it('rejects a handshake with protocol ' + JSON.stringify(badProtocol), function(done) { expectRejected({a: ACTIONS.handshake, protocol: badProtocol, protocolMinor: 2}, 'Invalid protocol', done); diff --git a/test/client/presence/doc-presence.js b/test/client/presence/doc-presence.js index 5cb4b5ef7..9e207cad4 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -601,6 +601,223 @@ describe('DocPresence', function() { connection1.send(message); }); + it('rejects a presence message for a different doc than its channel', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + expect(error.message).to.equal('Presence collection and doc ID do not match channel'); + done(); + }); + + var message = localPresence1._message(); + message.d = 'subtle-knife'; + message.v = 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message with a doc ID but no collection', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.c = null; + message.v = 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message with a collection but no doc ID', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.d = null; + message.v = 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message with an empty collection', function(done) { + var localPresence1 = connection1.getDocPresence('', 'northern-lights').create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.v = 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message with an empty doc ID', function(done) { + var localPresence1 = connection1.getDocPresence('books', '').create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.v = 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects presence on a doc the client cannot read', function(done) { + backend.use(backend.MIDDLEWARE_ACTIONS.readSnapshots, function(context, next) { + next(new Error('Forbidden')); + }); + + var localPresence1 = presence1.create('presence-1'); + localPresence1.submit({index: 1}, function(error) { + expect(error.message).to.equal('Forbidden'); + done(); + }); + }); + + it('ignores untyped presence on its channel', function(done) { + var connection3 = backend.connect(); + var untypedPresence3 = connection3.getPresence('books.northern-lights'); + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + untypedPresence3.subscribe.bind(untypedPresence3), + function(next) { + connection2.on('receive', function(request) { + if (request.data.src !== connection3.id) return; + localPresence1.submit({index: 1}, errorHandler(done)); + }); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.eql({index: 1}); + next(); + }); + connection3.send({ + a: 'p', ch: 'books.northern-lights', id: 'presence-1', p: {index: 9}, pv: 0, v: doc1.version + }); + } + ], done); + }); + + it('does not let untyped presence with the same ID ack its local presence', function(done) { + var connection3 = backend.connect(); + var untypedPresence3 = connection3.getPresence('books.northern-lights'); + var localPresence1 = presence1.create('presence-1'); + var hasReceivedOwnPresence = false; + + async.series([ + presence1.subscribe.bind(presence1), + untypedPresence3.subscribe.bind(untypedPresence3), + function(next) { + connection1.on('receive', function(request) { + if (request.data.src === connection1.id) hasReceivedOwnPresence = true; + if (request.data.src !== connection3.id) return; + localPresence1.submit({index: 1}, function(error) { + if (error) return done(error); + expect(hasReceivedOwnPresence).to.be.true; + next(); + }); + }); + connection3.send({ + a: 'p', ch: 'books.northern-lights', id: 'presence-1', p: {index: 9}, pv: 0, v: doc1.version + }); + } + ], done); + }); + + it('does not let typed presence from another client with the same ID ack its local presence', function(done) { + var connection3 = backend.connect(); + var presence3 = connection3.getDocPresence('books', 'northern-lights'); + var localPresence1 = presence1.create('presence-1'); + var hasReceivedOwnPresence = false; + + async.series([ + presence1.subscribe.bind(presence1), + presence3.subscribe.bind(presence3), + function(next) { + connection1.on('receive', function(request) { + if (request.data.src === connection1.id) hasReceivedOwnPresence = true; + if (request.data.src !== connection3.id) return; + localPresence1.submit({index: 1}, function(error) { + if (error) return done(error); + expect(hasReceivedOwnPresence).to.be.true; + next(); + }); + }); + connection3.send({ + a: 'p', + ch: 'books.northern-lights', + c: 'books', + d: 'northern-lights', + t: presenceTestType.type.uri, + id: 'presence-1', + p: {index: 9}, + pv: 0, + v: doc1.version + }); + } + ], done); + }); + + it('emits a forwarded error for a presence with the same ID on the presence, not its local presence', function(done) { + backend.use(backend.MIDDLEWARE_ACTIONS.sendPresence, function(context, next) { + next(new Error('uh-oh!')); + }); + + var connection3 = backend.connect(); + var doc3 = connection3.get('books', 'northern-lights'); + var localPresence3 = connection3.getDocPresence('books', 'northern-lights').create('presence-1'); + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', done); + + async.series([ + doc3.subscribe.bind(doc3), + presence1.subscribe.bind(presence1), + function(next) { + presence1.once('error', function(error) { + expect(error.message).to.equal('uh-oh!'); + next(); + }); + localPresence3.submit({index: 1}, errorHandler(done)); + } + ], done); + }); + + it('ignores presence for a doc whose collection and id join to the same channel', function(done) { + var connection3 = backend.connect(); + var notesDoc1 = connection1.get('notes', 'private.alice'); + var notesDoc3 = connection3.get('notes', 'private.alice'); + var diaryDoc2 = connection2.get('notes.private', 'alice'); + var localNotesPresence1 = connection1.getDocPresence('notes', 'private.alice').create('notes-presence'); + var localDiaryPresence2 = connection2.getDocPresence('notes.private', 'alice').create('diary-presence'); + var notesPresence3 = connection3.getDocPresence('notes', 'private.alice'); + + async.series([ + notesDoc1.create.bind(notesDoc1, 'Hello', presenceTestType.type.name), + diaryDoc2.create.bind(diaryDoc2, 'Dear diary', presenceTestType.type.name), + notesDoc3.subscribe.bind(notesDoc3), + notesPresence3.subscribe.bind(notesPresence3), + function(next) { + notesPresence3.once('receive', function(id) { + expect(id).to.equal('notes-presence'); + next(); + }); + localDiaryPresence2.submit({index: 1}, function(error) { + if (error) return done(error); + localNotesPresence1.submit({index: 1}, errorHandler(done)); + }); + } + ], done); + }); + it('only sends presence responses for the associated doc', function(done) { var localPresence1 = presence1.create('presence-1'); var localPresence2 = presence2.create('presence-2'); @@ -976,12 +1193,19 @@ describe('DocPresence', function() { var localPresence1 = presence1.create('presence-1'); var catchUpCount = 0; - function sendUntypedPresenceAtVersion(version) { - var message = localPresence1._message(); - message.d = null; - message.p = {index: 0}; - message.v = version; - connection1.send(message); + function receiveUntransformablePresenceAtVersion(version) { + connection2.handleMessage({ + a: ACTIONS.presence, + ch: 'books.northern-lights', + src: connection1.id, + id: 'presence-1', + p: {index: 0}, + pv: version, + c: 'books', + d: 'northern-lights', + v: version, + t: null + }); } async.series([ @@ -992,7 +1216,7 @@ describe('DocPresence', function() { catchUpCount++; if (catchUpCount === 1) next(); }); - sendUntypedPresenceAtVersion(0); + receiveUntransformablePresenceAtVersion(0); }, function(next) { doc2.once('op', function() { @@ -1006,7 +1230,7 @@ describe('DocPresence', function() { expect(error.code).to.equal('ERR_DOC_TYPE_NOT_RECOGNIZED'); next(); }); - sendUntypedPresenceAtVersion(1); + receiveUntransformablePresenceAtVersion(1); }, function(next) { doc2.once('op', function() { diff --git a/test/client/presence/presence.js b/test/client/presence/presence.js index f972327e2..4c5905a6f 100644 --- a/test/client/presence/presence.js +++ b/test/client/presence/presence.js @@ -685,6 +685,32 @@ describe('Presence', function() { ], done); }); + it('ignores presence from another client with the same ID as its local presence', function(done) { + var connection3 = backend.connect(); + var localPresence3 = connection3.getPresence('test-channel').create('presence-1'); + var localPresence1 = presence1.create('presence-1'); + var hasReceivedOwnPresence = false; + presence1.on('receive', function(id) { + done(new Error('Unexpected presence received: ' + id)); + }); + + async.series([ + presence1.subscribe.bind(presence1), + function(next) { + connection1.on('receive', function(request) { + if (request.data.src === connection1.id) hasReceivedOwnPresence = true; + if (request.data.src !== connection3.id) return; + localPresence1.submit({index: 1}, function(error) { + if (error) return done(error); + expect(hasReceivedOwnPresence).to.be.true; + next(); + }); + }); + localPresence3.submit({index: 9}, errorHandler(done)); + } + ], done); + }); + it('adds itself back onto the connection after a destroy and a resubscribe', function(done) { async.series([ presence1.destroy.bind(presence1), @@ -795,6 +821,27 @@ describe('Presence', function() { ], done); }); + it('emits an error for a presence with the same ID on the presence, not its local presence', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence2.create('presence-1'); + localPresence2.on('error', done); + + backend.use(backend.MIDDLEWARE_ACTIONS.sendPresence, function(context, next) { + next(new Error('uh-oh!')); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 3}, errorHandler(done)); + presence2.once('error', function(error) { + expect(error.message).to.equal('uh-oh!'); + next(); + }); + } + ], done); + }); + it('emits errors on the server instead of sending to the client', function(done) { var localPresence1 = presence1.create('presence-1'); From c3be55edd982b817196929b837f05fbc13b5a4dd Mon Sep 17 00:00:00 2001 From: Alec Gibson <12036746+alecgibson@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:28:43 +0000 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=90=9B=20Reject=20presence=20whose=20?= =?UTF-8?q?version=20is=20ahead=20of=20the=20doc?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes https://github.com/share/sharedb/issues/735 At the moment, the server relays a doc presence whose `v` is ahead of the doc unchanged, which lets one client make every subscriber fetch: 1. Submit presence with `v: 1e9` to a doc with 20 subscribers 2. Each subscriber calls `doc.fetch()`, since `v > doc.version` 3. The presence stays pending, so every later op fetches again This change has the `Agent` pass its tracked doc version through to the presence transform, which reads ops from the lower of that and `v`. The tracker can lag the database but never lead it, so a `v` past the last op read is ahead of the doc. Ops in that lag may have been deleted, though, so we check the snapshot version before rejecting. A leave reads ops even when its `v` matches the tracked version: no client resends `null` presence, so a stale one would never be corrected. Doc presence must now carry an integer `v`, and so must ops that send one. A rejected presence rolls the `Agent`'s cached request back to the newest earlier one that wasn't rejected, so the leave it replays on disconnect isn't rejected too. Presence sent without `c`/`d` skips all of this, since the server can't tell it from generic presence, but clients ignore it since #752. We deliberately don't bound presence that is far *behind*: a `fetch` with `v: 0` already makes the server read the whole op history. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- lib/agent.js | 76 +++- lib/backend.js | 47 ++- test/agent.js | 5 + test/client/presence/doc-presence.js | 533 +++++++++++++++++++++++++++ 4 files changed, 644 insertions(+), 17 deletions(-) diff --git a/lib/agent.js b/lib/agent.js index 998b22927..eb5ac2c84 100644 --- a/lib/agent.js +++ b/lib/agent.js @@ -53,6 +53,9 @@ function Agent(backend, stream) { // ops to transform presence if not needed this.latestDocVersionStreams = Object.create(null); this.latestDocVersions = Object.create(null); + // Presence requests whose transform is still in flight, each with the + // request it restores to presenceRequests if it's rejected + this._pendingPresenceTransforms = []; // We need to track this manually to make sure we don't reply to messages // after the stream was closed. @@ -449,9 +452,14 @@ Agent.prototype._checkRequest = function(request) { if (isInvalidVersion(request.v)) return 'Invalid version'; if (request.ts != null && typeof request.ts !== 'number') return 'Invalid timestamp'; + if (request.a === ACTIONS.op || request.a === ACTIONS.presence) { + if (request.v != null && !util.isInteger(request.v)) return 'Invalid version'; + } + if (request.a === ACTIONS.presence) { if (isInvalidName(request.id)) return 'Invalid presence ID'; if (isInvalidVersion(request.pv)) return 'Invalid presence version'; + if (isDocPresence(request) && request.v == null) return 'Invalid version'; } } else if ( request.a === ACTIONS.bulkFetch || @@ -864,6 +872,7 @@ Agent.prototype._broadcastPresence = function(presence, callback) { collection: presence.c }; var start = Date.now(); + var docVersionBeforeUpdate = util.dig(this.latestDocVersions, presence.c, presence.d); var subscriptionUpdater = presence.p === null ? this._unsubscribeDocVersion.bind(this) : @@ -875,33 +884,68 @@ Agent.prototype._broadcastPresence = function(presence, callback) { if (error) return callback(error); var requests = presenceRequests[presence.ch] || (presenceRequests[presence.ch] = Object.create(null)); var previousRequest = requests[presence.id]; - if (!previousRequest || previousRequest.pv < presence.pv) { - presenceRequests[presence.ch][presence.id] = presence; - } - - var transformer = function(agent, presence, callback) { - callback(null, presence); - }; + cacheNewerPresenceRequest(requests, presence); + var transform = {request: presence, fallbackRequest: previousRequest}; + agent._pendingPresenceTransforms.push(transform); var latestDocVersion = util.dig(agent.latestDocVersions, presence.c, presence.d); - var presenceIsUpToDate = presence.v === latestDocVersion; - if (!presenceIsUpToDate) { - transformer = backend.transformPresenceToLatestVersion.bind(backend); - } + if (latestDocVersion == null) latestDocVersion = docVersionBeforeUpdate; - transformer(agent, presence, function(error, presence) { - if (error) return callback(error); - var channel = agent._getPresenceChannel(presence.ch); - agent.backend.pubsub.publish([channel], presence, function(error) { + backend._transformPresenceToLatestVersion(agent, presence, latestDocVersion, function(error, latestPresence) { + agent._removePendingPresenceTransform(transform); + if (error) { + agent._rejectPresenceTransform(requests, transform); + return callback(error); + } + cacheNewerPresenceRequest(requests, presence); + agent._raisePendingPresenceFallbacks(presence); + var channel = agent._getPresenceChannel(latestPresence.ch); + agent.backend.pubsub.publish([channel], latestPresence, function(error) { if (error) return callback(error); backend.emit('timing', 'presence.broadcast', Date.now() - start, context); - callback(null, presence); + callback(null, latestPresence); }); }); }); }); }; +Agent.prototype._removePendingPresenceTransform = function(transform) { + var transforms = this._pendingPresenceTransforms; + var index = transforms.indexOf(transform); + if (index !== -1) transforms.splice(index, 1); +}; + +Agent.prototype._raisePendingPresenceFallbacks = function(acceptedRequest) { + var transforms = this._pendingPresenceTransforms; + for (var i = 0; i < transforms.length; i++) { + var request = transforms[i].request; + var fallbackRequest = transforms[i].fallbackRequest; + var isSamePresence = request.ch === acceptedRequest.ch && request.id === acceptedRequest.id; + var fallbackIsOlder = !fallbackRequest || fallbackRequest.pv < acceptedRequest.pv; + if (isSamePresence && request.pv > acceptedRequest.pv && fallbackIsOlder) { + transforms[i].fallbackRequest = acceptedRequest; + } + } +}; + +Agent.prototype._rejectPresenceTransform = function(requests, rejectedTransform) { + var rejectedRequest = rejectedTransform.request; + var fallbackRequest = rejectedTransform.fallbackRequest; + var transforms = this._pendingPresenceTransforms; + for (var i = 0; i < transforms.length; i++) { + if (transforms[i].fallbackRequest === rejectedRequest) transforms[i].fallbackRequest = fallbackRequest; + } + if (requests[rejectedRequest.id] !== rejectedRequest) return; + if (fallbackRequest) requests[rejectedRequest.id] = fallbackRequest; + else delete requests[rejectedRequest.id]; +}; + +function cacheNewerPresenceRequest(requests, request) { + var cachedRequest = requests[request.id]; + if (!cachedRequest || cachedRequest.pv < request.pv) requests[request.id] = request; +} + Agent.prototype._subscribeDocVersion = function(collection, id, callback) { if (!collection || !id) return callback(); diff --git a/lib/backend.js b/lib/backend.js index 17a409df1..834316a48 100644 --- a/lib/backend.js +++ b/lib/backend.js @@ -936,8 +936,12 @@ Backend.prototype._buildSnapshotFromOps = function(id, startingSnapshot, ops, ca }; Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, callback) { + this._transformPresenceToLatestVersion(agent, presence, null, callback); +}; + +Backend.prototype._transformPresenceToLatestVersion = function(agent, presence, trackedVersion, callback) { if (!presence.c || !presence.d) return callback(null, presence); - this.getOps(agent, presence.c, presence.d, presence.v, null, function(error, ops) { + this._getOpsSincePresence(agent, presence, trackedVersion, function(error, ops) { if (error) return callback(error); for (var i = 0; i < ops.length; i++) { var op = ops[i]; @@ -951,6 +955,47 @@ Backend.prototype.transformPresenceToLatestVersion = function(agent, presence, c }); }; +Backend.prototype._getOpsSincePresence = function(agent, presence, trackedVersion, callback) { + if (trackedVersion == null) return this._getOpsSincePresenceFromSnapshotVersion(agent, presence, callback); + var isLeave = presence.p === null; + if (!isLeave && presence.v === trackedVersion) return callback(null, []); + var backend = this; + this._getOpsSincePresenceFromVersion(agent, presence, trackedVersion, function(error, ops) { + if (error && presence.v > trackedVersion) { + return backend._getOpsSincePresenceFromSnapshotVersion(agent, presence, callback); + } + callback(error, ops); + }); +}; + +Backend.prototype._getOpsSincePresenceFromSnapshotVersion = function(agent, presence, callback) { + var backend = this; + var projection = this.projections[presence.c]; + var collection = (projection) ? projection.target : presence.c; + this.db.getSnapshot(collection, presence.d, {}, {agentCustom: agent.custom}, function(error, snapshot) { + if (error) return callback(error); + if (presence.v === snapshot.v) return callback(null, []); + backend._getOpsSincePresenceFromVersion(agent, presence, snapshot.v, callback); + }); +}; + +Backend.prototype._getOpsSincePresenceFromVersion = function(agent, presence, knownVersion, callback) { + var from = Math.min(presence.v, knownVersion); + this.getOps(agent, presence.c, presence.d, from, null, function(error, ops) { + if (error) return callback(error); + var versionAfterOps = ops.length ? ops[ops.length - 1].v + 1 : from; + if (presence.v > versionAfterOps) { + return callback(new ShareDBError( + ERROR_CODE.ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT, + 'Presence version newer than current snapshot' + )); + } + callback(null, ops.filter(function(op) { + return op.v >= presence.v; + })); + }); +}; + function pluckIds(snapshots) { var ids = []; for (var i = 0; i < snapshots.length; i++) { diff --git a/test/agent.js b/test/agent.js index 75574b636..80ed41cf5 100644 --- a/test/agent.js +++ b/test/agent.js @@ -144,6 +144,11 @@ describe('Agent', function() { }); }); + it('rejects an op with a non-integer version', function(done) { + var message = {a: ACTIONS.op, c: 'dogs', d: 'fido', v: 1.5, seq: 1, op: [{p: ['age'], na: 1}]}; + expectRejected(message, 'Invalid version', done); + }); + [unconvertible, ['dogs'], '__proto__'].forEach(function(badCollection) { it('rejects doc presence with collection ' + JSON.stringify(badCollection), function(done) { var message = {a: ACTIONS.presence, ch: 'dogs.fido', c: badCollection, d: 'fido', id: 'rex', pv: 0, v: 0}; diff --git a/test/client/presence/doc-presence.js b/test/client/presence/doc-presence.js index 9e207cad4..c1cf475ec 100644 --- a/test/client/presence/doc-presence.js +++ b/test/client/presence/doc-presence.js @@ -585,6 +585,539 @@ describe('DocPresence', function() { connection1.send(message); }); + it('rejects a presence message with a non-integer version', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.v = 1.5; + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + it('rejects a presence message without a version', function(done) { + var localPresence1 = presence1.create('presence-1'); + localPresence1.on('error', function(error) { + expect(error.code).to.eql('ERR_MESSAGE_BADLY_FORMED'); + done(); + }); + + var message = localPresence1._message(); + message.t = presenceTestType.type.uri; + connection1.send(message); + }); + + [ + {description: 'without doc coordinates', c: undefined, d: undefined}, + {description: 'for another doc', c: 'books', d: 'subtle-knife'} + ].forEach(function(test) { + it('ignores remote presence on its channel ' + test.description, function(done) { + sinon.spy(doc2, 'fetch'); + + [doc2.version, doc2.version + 1].forEach(function(version, index) { + var message = presence1.create('presence-' + index)._message(); + message.c = test.c; + message.d = test.d; + message.p = {index: 1}; + message.v = version; + message.t = presenceTestType.type.uri; + connection2.handleMessage(message); + }); + + doc2.whenNothingPending(function() { + expect(presence2.remotePresences).to.be.empty; + expect(presence2._remotePresenceInstances).to.be.empty; + expect(doc2.fetch).not.to.have.been.called; + done(); + }); + }); + }); + + [{index: 1}, null].forEach(function(value) { + it('rejects ' + JSON.stringify(value) + ' presence whose version is ahead of the doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + presence2.on('receive', function() { + done(new Error('Should not broadcast presence from the future')); + }); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.on('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + expect(connection1.agent.presenceRequests[presence1.channel]).not.to.have.property('presence-1'); + next(); + }); + + var message = localPresence1._message(); + message.p = value; + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + }); + + it('rejects ' + JSON.stringify(value) + ' presence ahead of a tracked doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + localPresence1.submit.bind(localPresence1, {index: 1}), + function(next) { + localPresence1.on('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + + var message = localPresence1._message(); + message.p = value; + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + }); + }); + + it('broadcasts null presence on disconnect after rejecting presence ahead of the doc', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + + var message = localPresence1._message(); + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('keeps newer presence cached when an earlier presence ahead of the doc is rejected', function(done) { + var localPresence1 = presence1.create('presence-1'); + var releaseGetOps; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + var getOps = backend.db.getOps; + sinon.stub(backend.db, 'getOps').callsFake(function() { + var args = arguments; + releaseGetOps = function() { + getOps.apply(backend.db, args); + }; + next(); + }); + + var message = localPresence1._message(); + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + }, + function(next) { + backend.db.getOps.restore(); + localPresence1.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 2}); + next(); + }); + }, + function(next) { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + next(); + }); + releaseGetOps(); + }, + function(next) { + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + [[0, 1], [1, 0, 2]].forEach(function(releaseOrder) { + it('restores accepted presence after rejecting presences in order ' + releaseOrder, function(done) { + var localPresence1 = presence1.create('presence-1'); + var heldGetOps = []; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + var getOps = backend.db.getOps; + sinon.stub(backend.db, 'getOps').callsFake(function() { + var args = arguments; + heldGetOps.push(function() { + getOps.apply(backend.db, args); + }); + if (heldGetOps.length === releaseOrder.length) next(); + }); + + for (var i = 0; i < releaseOrder.length; i++) { + var message = localPresence1._message(); + message.v = doc1.version + 1; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + }, + function(next) { + backend.db.getOps.restore(); + async.eachSeries(releaseOrder, function(heldGetOpsIndex, nextRelease) { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + nextRelease(); + }); + heldGetOps[heldGetOpsIndex](); + }, next); + }, + function(next) { + var cachedRequest = connection1.agent.presenceRequests[presence1.channel]['presence-1']; + expect(cachedRequest.v).to.equal(doc1.version); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + }); + + it('accepts up-to-date presence when the agent\'s tracked doc version lags behind', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + sinon.spy(backend.db, 'getSnapshot'); + localPresence1.submit({index: 12}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(doc2.version).to.eql(2); + expect(presence).to.eql({index: 12}); + expect(backend.db.getSnapshot).not.to.have.been.called; + next(); + }); + } + ], done); + }); + + [true, false].forEach(function(hasEarlierPresence) { + [['older', 'newer'], ['newer', 'older']].forEach(function(releaseOrder) { + var title = 'keeps older accepted presence cached when a newer one is rejected, in order ' + releaseOrder + + (hasEarlierPresence ? ', after earlier presence' : ', with no earlier presence'); + + it(title, function(done) { + var localPresence1 = presence1.create('presence-1'); + var heldGetOps = {}; + var heldGetOpsCount = 0; + var olderMessage; + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + if (!hasEarlierPresence) return next(); + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + function(next) { + olderMessage = localPresence1._message(); + olderMessage.p = {index: 1}; + olderMessage.v = doc1.version - 1; + olderMessage.t = presenceTestType.type.uri; + var newerMessage = localPresence1._message(); + newerMessage.p = {index: 1}; + newerMessage.v = doc1.version + 1; + newerMessage.t = presenceTestType.type.uri; + + var getOps = backend.db.getOps; + sinon.stub(backend.db, 'getOps').callsFake(function() { + var args = arguments; + var fromVersion = args[2]; + var age = fromVersion === olderMessage.v ? 'older' : 'newer'; + heldGetOps[age] = function() { + getOps.apply(backend.db, args); + }; + if (++heldGetOpsCount === 2) next(); + }); + + connection1.send(newerMessage); + connection1.send(olderMessage); + }, + function(next) { + backend.db.getOps.restore(); + async.eachSeries(releaseOrder, function(age, nextRelease) { + if (age === 'older') { + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 1}); + nextRelease(); + }); + } else { + localPresence1.once('error', function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + nextRelease(); + }); + } + heldGetOps[age](); + }, next); + }, + function(next) { + var cachedRequest = connection1.agent.presenceRequests[presence1.channel]['presence-1']; + expect(cachedRequest.pv).to.equal(olderMessage.pv); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + }); + }); + + it('clears presence on disconnect when the agent\'s tracked doc version lags behind', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence1.subscribe.bind(presence1), + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc2.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc1.once('op', function() { + next(); + }); + }, + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + presence2.once('receive', function(id, presence) { + expect(doc2.version).to.equal(2); + expect(id).to.equal('presence-1'); + expect(presence).to.be.null; + next(); + }); + connection1.close(); + } + ], done); + }); + + it('accepts up-to-date presence when the agent\'s tracked doc version lags behind deleted ops', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + localPresence1.submit({index: 12}, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 12}); + next(); + }); + } + ], done); + }); + + it( + 'transforms presence by surviving ops when the agent\'s tracked doc version lags behind deleted ops', + function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + doc1.submitOp({index: 5, value: 'ern'}, errorHandler(done)); + doc2.once('op', function() { + next(); + }); + }, + function(next) { + doc2.submitOp({index: 0, value: 'The '}, errorHandler(done)); + doc1.once('op', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', 0, 2, null), + function(next) { + connection1.agent.latestDocVersions.books['northern-lights'] = 1; + presence2.once('receive', function(id, presence) { + expect(presence).to.eql({index: 16}); + next(); + }); + + var message = localPresence1._message(); + message.p = {index: 12}; + message.v = 2; + message.t = presenceTestType.type.uri; + connection1.send(message); + } + ], done); + } + ); + + it('rejects presence ahead of the doc when transformed without a tracked version', function(done) { + var presence = { + c: 'books', + d: 'northern-lights', + p: {index: 1}, + v: doc1.version + 1, + t: presenceTestType.type.uri + }; + backend.transformPresenceToLatestVersion(connection1.agent, presence, function(error) { + expect(error.code).to.equal('ERR_OP_VERSION_NEWER_THAN_CURRENT_SNAPSHOT'); + done(); + }); + }); + + it('broadcasts null presence on an up-to-date doc whose ops have been deleted', function(done) { + var localPresence1 = presence1.create('presence-1'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + sinon.spy(backend.db, 'getSnapshot'); + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(presence).to.be.null; + expect(backend.db.getSnapshot).not.to.have.been.called; + next(); + }); + } + ], done); + }); + + it('broadcasts null presence from a second local presence on a doc whose ops have been deleted', function(done) { + var localPresence1 = presence1.create('presence-1'); + var localPresence2 = presence1.create('presence-2'); + + async.series([ + presence2.subscribe.bind(presence2), + function(next) { + localPresence1.submit({index: 1}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + localPresence2.submit({index: 2}, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + backend.db.deleteOps.bind(backend.db, 'books', 'northern-lights', null, null, null), + function(next) { + localPresence1.submit(null, errorHandler(done)); + presence2.once('receive', function() { + next(); + }); + }, + function(next) { + sinon.spy(backend.db, 'getSnapshot'); + localPresence2.submit(null, errorHandler(done)); + presence2.once('receive', function(id, presence) { + expect(id).to.equal('presence-2'); + expect(presence).to.be.null; + expect(backend.db.getSnapshot).to.have.been.called; + next(); + }); + } + ], done); + }); + it('rejects a presence message without an ID', function(done) { var localPresence1 = presence1.create('presence-1'); // Have to catch the error on the Presence instance, because obviously