🐛 Only read known query results after 'query' middleware

docs/middleware/access-control.md

@@ -0,0 +1,77 @@
+---
+title: Access control
+nav_order: 4
+layout: default
+parent: Middleware
+---
+
+# Access control
+{: .no_toc }
+
+1. TOC
+{:toc}
+
+ShareDB has no built-in access control. Instead, you enforce it in [middleware]({{ site.baseurl }}{% link middleware/registration.md %}).
+
+A client can read a document in two forms: as a snapshot, or as the ops that build it. These go through different middleware, so **read access control must be enforced in both [`'readSnapshots'`]({{ site.baseurl }}{% link middleware/actions.md %}#readsnapshots) and [`'op'`]({{ site.baseurl }}{% link middleware/actions.md %}#op) middleware**.
+
+{: .warn }
+The client chooses which form it asks for. Checking only `'readSnapshots'` still lets a client read a document's `create` data and its entire history by asking for its ops from version `0`.
+
+## Which middleware sees what
+
+| Data sent to the client | Middleware |
+|---|---|
+| Snapshots: fetching or subscribing to a doc the client doesn't have yet, bulk fetches and subscribes by id, query results (including docs added to a subscribed query's results later), resubscribing to a query after reconnecting, [`fetchSnapshot()`]({{ site.baseurl }}{% link api/connection.md %}#fetchsnapshot) and [`fetchSnapshotByTimestamp()`]({{ site.baseurl }}{% link api/connection.md %}#fetchsnapshotbytimestamp) | [`'readSnapshots'`]({{ site.baseurl }}{% link middleware/actions.md %}#readsnapshots) |
+| Ops: fetching or subscribing to a doc from a version, bulk fetches and subscribes from versions, resubscribing to a query after reconnecting, ops on subscribed docs and query results as they happen, and ops the client missed, sent back after it submits | [`'op'`]({{ site.baseurl }}{% link middleware/actions.md %}#op) |
+| Queries -- this runs before the query, and doesn't see its results | [`'query'`]({{ site.baseurl }}{% link middleware/actions.md %}#query) |
+| Presence | [`'sendPresence'`]({{ site.baseurl }}{% link middleware/actions.md %}#sendpresence) |
+| Writes | See [op submission]({{ site.baseurl }}{% link middleware/op-submission.md %}) |
+
+## Example
+
+```js
+function canRead(agent, collection, id) {
+  // Your own rule, e.g. checking against agent.custom.userId
+}
+
+function canQuery(agent, collection, query) {
+  // Your own rule, e.g. only allowing queries scoped to agent.custom.userId
+}
+
+backend.use('readSnapshots', (context, next) => {
+  for (const snapshot of context.snapshots) {
+    if (!canRead(context.agent, context.collection, snapshot.id)) {
+      context.rejectSnapshotRead(snapshot, new Error('Forbidden'))
+    }
+  }
+  next()
+})
+
+backend.use('op', (context, next) => {
+  if (!canRead(context.agent, context.collection, context.id)) {
+    return next(new Error('Forbidden'))
+  }
+  next()
+})
+
+backend.use('query', (context, next) => {
+  if (!canQuery(context.agent, context.collection, context.query)) {
+    return next(new Error('Forbidden'))
+  }
+  next()
+})
+```
+
+In a bulk fetch or subscribe, `context.rejectSnapshotRead()` rejects just that snapshot, and the rest of the request still succeeds. Queries can't partially succeed: if any of a query's snapshots is rejected, the whole query fails, and a subscribed query drops that update. Use `'query'` middleware to keep queries to documents the client can read -- for example, by narrowing `context.query`, whose shape depends on your [database adapter]({{ site.baseurl }}{% link adapters/database.md %}) -- and treat `'readSnapshots'` as the backstop.
+
+Narrowing a query only limits its results. A client resubscribing to a query after reconnecting names the documents it already has, and reading those is checked only by `'readSnapshots'` and `'op'`.
+
+There is no per-op equivalent of `rejectSnapshotRead()`: rejecting an op fails the whole fetch or subscribe that read it, including a query resubscribe. So if a client loses access to one of a subscribed query's documents while it's offline, the query errors when it reconnects. A live op that `'op'` rejects as it arrives over pub/sub is logged and dropped instead: the client's doc doesn't error, but stops updating until its next fetch or resubscribe, which is then rejected.
+
+{: .warn }
+`'op'` runs once for every op read, so a client catching up from version `0` triggers it for every op in the document's history. Keep the check cheap: for example, by caching permissions on `agent.custom`.
+
+## Projections
+
+For a [projection]({{ site.baseurl }}{% link projections.md %}), `context.collection` in `'readSnapshots'` and `'op'` is the projection's target collection, and the snapshot or op has already been projected.

docs/middleware/actions.md

@@ -137,7 +137,10 @@ This action has these additional `context` properties:
 
 ## `'readSnapshots'`
 
-One or more snapshots were loaded from the database for a fetch or subscribe.
+One or more snapshots were loaded from the database for a fetch, subscribe or query.
+
+{: .warn }
+Ops sent to the client do not pass through this action. Read access control also needs [`'op'`](#op) middleware. See [Access control]({{ site.baseurl }}{% link middleware/access-control.md %}).
 
 This action has these additional `context` properties:
 
@@ -158,7 +161,10 @@ This action has these additional `context` properties:
 
 ## `'op'`
 
-An operation was loaded from the database.
+An operation is being read for a client, from the database or over [pub/sub]({{ site.baseurl }}{% link pub-sub.md %}), usually to send to it.
+
+{: .warn }
+Snapshots sent to the client do not pass through this action. Read access control also needs [`'readSnapshots'`](#readsnapshots) middleware. See [Access control]({{ site.baseurl }}{% link middleware/access-control.md %}).
 
 This action has these additional `context` properties:
 

docs/middleware/index.md

@@ -10,6 +10,5 @@ has_children: true
 
 Middleware enables consumers to hook into the ShareDB server pipeline. Objects can be asynchronously manipulated as they flow through ShareDB.
 
-<!-- TODO: Link to an auth example -->
 <!-- TODO: Link to a metadata example -->
-This can be particularly useful for authentication, or adding metadata.
+This can be particularly useful for authentication, [access control]({{ site.baseurl }}{% link middleware/access-control.md %}), or adding metadata.

docs/projections.md

@@ -15,6 +15,9 @@ Once configured, the projected collection looks just like a real collection -- e
 
 Operations on the projected collection work, but only a small portion of the data can be seen and altered.
 
+{: .warn }
+Projections restrict fields, but they aren't access control on their own. Reads through a projection still run [`'readSnapshots'`]({{ site.baseurl }}{% link middleware/actions.md %}#readsnapshots) and [`'op'`]({{ site.baseurl }}{% link middleware/actions.md %}#op) middleware, with `collection` set to the target collection. See [Access control]({{ site.baseurl }}{% link middleware/access-control.md %}).
+
 ## Usage
 
 Projections are configured using [`backend.addProjection()`]({{ site.baseurl }}{% link api/backend.md %}#addprojection). For example, imagine we have a collection `users` with lots of information that should not be leaked. To add a projection `names`, which only has access to the `firstName` and `lastName` properties on a user:

lib/agent.js

@@ -1,3 +1,4 @@
+var async = require('async');
 var hat = require('hat');
 var ShareDBError = require('./error');
 var logger = require('./logger');
@@ -562,42 +563,51 @@ Agent.prototype._querySubscribe = function(queryId, collection, query, options,
   // Subscribe to a query. The client is sent the query results and its
   // notified whenever there's a change
   var agent = this;
-  var wait = 1;
-  var message;
-  function finish(err) {
+  this.backend.querySubscribe(this, collection, query, options, function(err, emitter, results, extra) {
     if (err) return callback(err);
-    if (--wait) return;
-    callback(null, message);
-  }
-  if (options.fetch) {
-    wait++;
-    this.backend.fetchBulk(this, collection, options.fetch, function(err, snapshotMap) {
-      if (err) return finish(err);
-      message = getMapResult(snapshotMap);
-      finish();
+    agent._fetchKnownResults(collection, options, function(err, message) {
+      if (err) {
+        emitter.destroy();
+        return callback(err);
+      }
+      if (agent.closed) return emitter.destroy();
+
+      agent._subscribeToQuery(emitter, queryId, collection, query);
+      if (results) return callback(null, {data: getResultsData(results), extra: extra});
+      // No results are returned when ids are passed in as an option. Instead,
+      // want to re-poll the entire query once we've established listeners to
+      // emit any diff in results
+      emitter.queryPoll(function(err) {
+        if (err) {
+          emitter.destroy();
+          if (agent.subscribedQueries[queryId] === emitter) delete agent.subscribedQueries[queryId];
+        }
+        callback(err, message);
+      });
     });
-  }
-  if (options.fetchOps) {
-    wait++;
-    this._fetchBulkOps(collection, options.fetchOps, finish);
-  }
-  this.backend.querySubscribe(this, collection, query, options, function(err, emitter, results, extra) {
-    if (err) return finish(err);
-    if (agent.closed) return emitter.destroy();
-
-    agent._subscribeToQuery(emitter, queryId, collection, query);
-    // No results are returned when ids are passed in as an option. Instead,
-    // want to re-poll the entire query once we've established listeners to
-    // emit any diff in results
-    if (!results) {
-      emitter.queryPoll(finish);
-      return;
+  });
+};
+
+Agent.prototype._fetchKnownResults = function(collection, options, callback) {
+  var agent = this;
+  async.parallel({
+    snapshots: function(cb) {
+      if (!options.fetch) return cb();
+      agent.backend.fetchBulk(agent, collection, options.fetch, function(err, snapshotMap) {
+        if (err) return cb(err);
+        var readSnapshotsError = getReadSnapshotsError(snapshotMap);
+        if (readSnapshotsError) return cb(readSnapshotsError);
+        cb(null, getMapResult(snapshotMap));
+      });
+    },
+    ops: function(cb) {
+      if (!options.fetchOps) return cb();
+      agent.backend.getOpsBulk(agent, collection, options.fetchOps, null, cb);
     }
-    message = {
-      data: getResultsData(results),
-      extra: extra
-    };
-    finish();
+  }, function(err, results) {
+    if (err) return callback(err);
+    if (results.ops) agent._sendOpsBulk(collection, results.ops);
+    callback(null, results.snapshots);
   });
 };
 
@@ -636,6 +646,20 @@ function getMapResult(snapshotMap) {
   return {data: data};
 }
 
+function getReadSnapshotsError(snapshotMap) {
+  var idToError = null;
+  for (var id in snapshotMap) {
+    var error = snapshotMap[id].error;
+    if (!error) continue;
+    if (!idToError) idToError = Object.create(null);
+    idToError[id] = error;
+  }
+  if (!idToError) return;
+  var err = new ShareDBError(ERROR_CODE.ERR_SNAPSHOT_READS_REJECTED);
+  err.idToError = idToError;
+  return err;
+}
+
 function getSnapshotData(snapshot) {
   var data = {
     v: snapshot.v,

lib/query-emitter.js

@@ -306,11 +306,9 @@ QueryEmitter.prototype.queryPollDoc = function(id, callback) {
   });
 };
 
-// Clients must assign each of these functions synchronously after constructing
-// an instance of QueryEmitter. The instance is subscribed to an op stream at
-// construction time, and does not buffer emitted events. Diff events assume
-// all messages are received and applied in order, so it is critical that none
-// are dropped.
+// Clients must assign each of these functions before calling _open(). The op
+// streams buffer emitted events until then. Diff events assume all messages
+// are received and applied in order, so it is critical that none are dropped.
 QueryEmitter.prototype.onError =
   QueryEmitter.prototype.onDiff =
   QueryEmitter.prototype.onExtra =

test/agent.js

@@ -7,6 +7,7 @@ var ACTIONS = require('../lib/message-actions').ACTIONS;
 var Connection = require('../lib/client/connection');
 var protocol = require('../lib/protocol');
 var LegacyConnection = require('sharedb-legacy/lib/client').Connection;
+var util = require('./util');
 
 describe('Agent', function() {
   var backend;
@@ -187,4 +188,141 @@ describe('Agent', function() {
       });
     });
   });
+
+  describe('query subscribe with known results', function() {
+    var connection;
+
+    beforeEach(function(done) {
+      connection = backend.connect();
+      connection.get('dogs', 'fido').create({name: 'Fido'}, done);
+    });
+
+    function resubscribe(knownResults, callback) {
+      connection.on('receive', function(request) {
+        if (request.data.a === ACTIONS.querySubscribe) callback(request.data);
+      });
+      connection.send({
+        a: ACTIONS.querySubscribe,
+        id: 1,
+        c: 'dogs',
+        q: {},
+        o: {pollDebounce: 0, pollInterval: 50},
+        r: knownResults
+      });
+    }
+
+    it('does not read known results when the query middleware rejects it', function(done) {
+      backend.use('query', function(context, next) {
+        next(new Error('Forbidden'));
+      });
+      sinon.spy(backend.db, 'getOpsBulk');
+      sinon.spy(backend.db, 'getSnapshotBulk');
+      resubscribe([['fido', 0], ['spot', null]], function(reply) {
+        expect(reply.error).to.have.property('message', 'Forbidden');
+        expect(backend.db.getOpsBulk).not.to.have.been.called;
+        expect(backend.db.getSnapshotBulk).not.to.have.been.called;
+        done();
+      });
+    });
+
+    it('does not stay subscribed if fetching known results fails', function(done) {
+      backend.use('op', function(context, next) {
+        next(new Error('Forbidden'));
+      });
+      resubscribe([['fido', 0], ['spot', null]], function(reply) {
+        expect(reply.error).to.have.property('message', 'Forbidden');
+        expect(connection.agent.subscribedQueries).to.be.empty;
+        expect(backend.pubsub.streamsCount).to.equal(0);
+        done();
+      });
+    });
+
+    it('does not stay subscribed if one of its known snapshots is rejected', function(done) {
+      backend.use('readSnapshots', function(context, next) {
+        context.snapshots.forEach(function(snapshot) {
+          if (snapshot.id === 'spot') context.rejectSnapshotRead(snapshot, new Error('Forbidden'));
+        });
+        next();
+      });
+      resubscribe([['fido', null], ['spot', null]], function(reply) {
+        expect(reply.error).to.have.property('code', 'ERR_SNAPSHOT_READS_REJECTED');
+        expect(connection.agent.subscribedQueries).to.be.empty;
+        expect(backend.pubsub.streamsCount).to.equal(0);
+        done();
+      });
+    });
+
+    it('does not send known results\' ops if reading their snapshots fails', function(done) {
+      backend.use('readSnapshots', function(context, next) {
+        next(new Error('Forbidden'));
+      });
+      var sentOps = [];
+      backend.on('send', function(agent, message) {
+        if (message.a === ACTIONS.op) sentOps.push(message);
+      });
+      var getOpsBulk = backend.getOpsBulk;
+      var finishGettingOps;
+      sinon.stub(backend, 'getOpsBulk').callsFake(function(agent, collection, fromMap, toMap, callback) {
+        finishGettingOps = function() {
+          getOpsBulk.call(backend, agent, collection, fromMap, toMap, function(error, opsMap) {
+            callback(error, opsMap);
+            expect(sentOps).to.be.empty;
+            done();
+          });
+        };
+      });
+      resubscribe([['fido', 0], ['spot', null]], function(reply) {
+        expect(reply.error).to.have.property('message', 'Forbidden');
+        finishGettingOps();
+      });
+    });
+
+    it('does not stay subscribed if re-polling the query fails', function(done) {
+      var clock = util.useFakeTimers();
+      connection.get('dogs', 'rex').create({name: 'Rex'}, function(error) {
+        if (error) return done(error);
+        backend.use('readSnapshots', function(context, next) {
+          var forbidden = context.snapshots.some(function(snapshot) {
+            return snapshot.id === 'rex';
+          });
+          next(forbidden ? new Error('Forbidden') : null);
+        });
+        resubscribe([['fido', 0], ['spot', null]], function(reply) {
+          expect(reply.error).to.have.property('message', 'Forbidden');
+          expect(connection.agent.subscribedQueries).to.be.empty;
+          expect(backend.pubsub.streamsCount).to.equal(0);
+          sinon.spy(backend.db, 'queryPoll');
+          clock.tick(1000);
+          expect(backend.db.queryPoll).not.to.have.been.called;
+          done();
+        });
+      });
+    });
+
+    it('delivers an op committed while fetching known results', function(done) {
+      var writer = backend.connect().get('dogs', 'fido');
+      connection.createSubscribeQuery('dogs', {}, null, function(error, results) {
+        if (error) return done(error);
+        var fido = results[0];
+        fido.on('op', function() {
+          expect(fido.data).to.eql({name: 'Rex'});
+          done();
+        });
+        var getOpsBulk = backend.db.getOpsBulk;
+        sinon.stub(backend.db, 'getOpsBulk').callsFake(function(collection, fromMap, toMap, options, callback) {
+          backend.db.getOpsBulk.restore();
+          getOpsBulk.call(backend.db, collection, fromMap, toMap, options, function(error, opsMap) {
+            writer.fetch(function(error) {
+              if (error) return done(error);
+              writer.submitOp({p: ['name'], od: 'Fido', oi: 'Rex'}, function(error) {
+                callback(error, opsMap);
+              });
+            });
+          });
+        });
+        connection.close();
+        backend.connect(connection);
+      });
+    });
+  });
 });