Merge pull request #331 from share/readSnapshots-specific-errors

Add ability for "readSnapshots" middleware to reject reads of specific snapshots for bulk fetch/subscribe

Author: ericyhwang

README.md

@@ -172,6 +172,8 @@ Register a new middleware.
     * `stream`: The duplex Stream provided to `share.listen` (for 'connect')
     * `query`: The query object being handled (for 'query')
     * `snapshots`: Array of retrieved snapshots (for 'readSnapshots')
+    * `rejectSnapshotRead(snapshot, error)`: Reject a specific snapshot read (for 'readSnapshots')
+      - `rejectSnapshotReadSilent(snapshot, errorMessage)`: As above, but causes the ShareDB client to treat it as a silent rejection, not passing the error back to user code.
     * `data`: Received client message (for 'receive')
     * `request`: Client message being replied to (for 'reply')
     * `reply`: Reply to be sent to the client (for 'reply')
@@ -601,21 +603,28 @@ An `Agent` is the representation of a client's `Connection` state on the server.
 The `Agent` will be made available in all [middleware](#middlewares) requests. The `agent.custom` field is an object that can be used for storing arbitrary information for use in middleware. For example:
 
 ```javascript
-backend.useMiddleware('connect', function (request, callback) {
+backend.useMiddleware('connect', (request, callback) => {
   // Best practice to clone to prevent mutating the object after connection.
   // You may also want to consider a deep clone, depending on the shape of request.req.
   Object.assign(request.agent.custom, request.req);
   callback();
 });
 
-backend.useMiddleware('readSnapshots', function (request, callback) {
-  var connectionInfo = request.agent.custom;
-  var snapshots = request.snapshots;
+backend.useMiddleware('readSnapshots', (request, callback) => {
+  const connectionInfo = request.agent.custom;
+  const snapshots = request.snapshots;
 
-  // Use the information provided at connection to determine if a user can access snapshots.
+  // Use the information provided at connection to determine if a user can access the snapshots.
   // This should also be checked when fetching and submitting ops.
-  if (!userCanAccessSnapshots(connectionInfo, snapshots)) {
-    return callback(new Error('Authentication error'));
+  if (!userCanAccessCollection(connectionInfo, request.collection)) {
+    return callback(new Error('Not allowed to access collection ' + request.collection));
+  }
+  // Check each snapshot individually.
+  for (const snapshot of snapshots) {
+    if (!userCanAccessSnapshot(connectionInfo, request.collection, snapshot)) {
+      request.rejectSnapshotRead(snapshot,
+        new Error('Not allowed to access snapshot in ' request.collection));
+    }
   }
 
   callback();
@@ -625,10 +634,10 @@ backend.useMiddleware('readSnapshots', function (request, callback) {
 // potentially making some database request to check which documents they can access, or which
 // roles they have, etc. If doing this asynchronously, make sure you call backend.connect
 // after the permissions have been fetched.
-var connectionInfo = getUserPermissions();
+const connectionInfo = getUserPermissions();
 // Pass info in as the second argument. This will be made available as request.req in the
 // 'connection' middleware.
-var connection = backend.connect(null, connectionInfo);
+const connection = backend.connect(null, connectionInfo);
 ```
 
 ### Logging

lib/agent.js

@@ -424,7 +424,7 @@ Agent.prototype._querySubscribe = function(queryId, collection, query, options,
     wait++;
     this.backend.fetchBulk(this, collection, options.fetch, function(err, snapshotMap) {
       if (err) return finish(err);
-      message = {data: getMapData(snapshotMap)};
+      message = getMapResult(snapshotMap);
       finish();
     });
   }
@@ -463,12 +463,20 @@ function getResultsData(results) {
   return items;
 }
 
-function getMapData(snapshotMap) {
+function getMapResult(snapshotMap) {
   var data = {};
   for (var id in snapshotMap) {
-    data[id] = getSnapshotData(snapshotMap[id]);
+    var mapValue = snapshotMap[id];
+    // fetchBulk / subscribeBulk map data can have either a Snapshot or an object
+    // `{error: Error | string}` as a value.
+    if (mapValue.error) {
+      // Transform errors to serialization-friendly objects.
+      data[id] = {error: getReplyErrorObject(mapValue.error)};
+    } else {
+      data[id] = getSnapshotData(mapValue);
+    }
   }
-  return data;
+  return {data: data};
 }
 
 function getSnapshotData(snapshot) {
@@ -517,8 +525,15 @@ Agent.prototype._fetchOps = function(collection, id, version, callback) {
 Agent.prototype._fetchBulk = function(collection, versions, callback) {
   if (Array.isArray(versions)) {
     this.backend.fetchBulk(this, collection, versions, function(err, snapshotMap) {
-      if (err) return callback(err);
-      callback(null, {data: getMapData(snapshotMap)});
+      if (err) {
+        return callback(err);
+      }
+      if (snapshotMap) {
+        var result = getMapResult(snapshotMap);
+        callback(null, result);
+      } else {
+        callback();
+      }
     });
   } else {
     this._fetchBulkOps(collection, versions, callback);
@@ -565,15 +580,18 @@ Agent.prototype._subscribeBulk = function(collection, versions, callback) {
   // See _subscribe() above. This function's logic should match but in bulk
   var agent = this;
   this.backend.subscribeBulk(this, collection, versions, function(err, streams, snapshotMap, opsMap) {
-    if (err) return callback(err);
+    if (err) {
+      return callback(err);
+    }
     if (opsMap) {
       agent._sendOpsBulk(collection, opsMap);
     }
     for (var id in streams) {
       agent._subscribeToStream(collection, id, streams[id]);
     }
     if (snapshotMap) {
-      callback(null, {data: getMapData(snapshotMap)});
+      var result = getMapResult(snapshotMap);
+      callback(null, result);
     } else {
       callback();
     }

lib/backend.js

@@ -12,6 +12,7 @@ var ShareDBError = require('./error');
 var Snapshot = require('./snapshot');
 var StreamSocket = require('./stream-socket');
 var SubmitRequest = require('./submit-request');
+var ReadSnapshotsRequest = require('./read-snapshots-request');
 
 var ERROR_CODE = ShareDBError.CODES;
 
@@ -252,13 +253,21 @@ Backend.prototype._sanitizeSnapshots = function(agent, projection, collection, s
     }
   }
 
-  var request = {
-    collection: collection,
-    snapshots: snapshots,
-    snapshotType: snapshotType
-  };
+  var request = new ReadSnapshotsRequest(collection, snapshots, snapshotType);
 
-  this.trigger(this.MIDDLEWARE_ACTIONS.readSnapshots, agent, request, callback);
+  this.trigger(this.MIDDLEWARE_ACTIONS.readSnapshots, agent, request, function(err) {
+    if (err) return callback(err);
+    // Handle "partial rejection" - "readSnapshots" middleware functions can use
+    // `request.rejectSnapshotRead(snapshot, error)` to reject the read of a specific snapshot.
+    if (request.hasSnapshotRejection()) {
+      err = request.getReadSnapshotsError();
+    }
+    if (err) {
+      callback(err);
+    } else {
+      callback();
+    }
+  });
 };
 
 Backend.prototype._getSnapshotProjection = function(db, projection) {
@@ -382,6 +391,18 @@ Backend.prototype.fetch = function(agent, index, id, options, callback) {
   });
 };
 
+/**
+ * Map of document id to Snapshot or error object.
+ * @typedef {{ [id: string]: Snapshot | { error: Error | string } }} SnapshotMap
+ */
+
+/**
+ * @param {Agent} agent
+ * @param {string} index
+ * @param {string[]} ids
+ * @param {*} options
+ * @param {(err?: Error | string, snapshotMap?: SnapshotMap) => void} callback
+ */
 Backend.prototype.fetchBulk = function(agent, index, ids, options, callback) {
   if (typeof options === 'function') {
     callback = options;
@@ -410,9 +431,18 @@ Backend.prototype.fetchBulk = function(agent, index, ids, options, callback) {
       snapshots,
       backend.SNAPSHOT_TYPES.current,
       function(err) {
-        if (err) return callback(err);
+        if (err) {
+          if (err.code === ERROR_CODE.ERR_SNAPSHOT_READS_REJECTED) {
+            for (var docId in err.idToError) {
+              snapshotMap[docId] = {error: err.idToError[docId]};
+            }
+            err = undefined;
+          } else {
+            snapshotMap = undefined;
+          }
+        }
         backend.emit('timing', 'fetchBulk', Date.now() - start, request);
-        callback(null, snapshotMap);
+        callback(err, snapshotMap);
       });
   });
 };
@@ -471,6 +501,26 @@ Backend.prototype.subscribe = function(agent, index, id, version, options, callb
   });
 };
 
+/**
+ * Map of document id to pubsub stream.
+ * @typedef {{ [id: string]: Stream }} StreamMap
+ */
+/**
+ * Map of document id to array of ops for the doc.
+ * @typedef {{ [id: string]: Op[] }} OpsMap
+ */
+
+/**
+ * @param {Agent} agent
+ * @param {string} index
+ * @param {string[]} versions
+ * @param {(
+ *   err?: Error | string | null,
+ *   streams?: StreamMap,
+ *   snapshotMap?: SnapshotMap | null
+ *   opsMap?: OpsMap
+ * ) => void} callback
+ */
 Backend.prototype.subscribeBulk = function(agent, index, versions, callback) {
   var start = Date.now();
   var projection = this.projections[index];
@@ -501,11 +551,21 @@ Backend.prototype.subscribeBulk = function(agent, index, versions, callback) {
       // If an array of ids, get current snapshots
       backend.fetchBulk(agent, index, ids, function(err, snapshotMap) {
         if (err) {
+          // Full error, destroy all streams.
           destroyStreams(streams);
-          return callback(err);
+          streams = undefined;
+          snapshotMap = undefined;
+        }
+        for (var docId in snapshotMap) {
+          // The doc id could map to an object `{error: Error | string}`, which indicates that
+          // particular snapshot's read was rejected. Destroy the streams fur such docs.
+          if (snapshotMap[docId].error) {
+            streams[docId].destroy();
+            delete streams[docId];
+          }
         }
         backend.emit('timing', 'subscribeBulk.snapshot', Date.now() - start, request);
-        callback(null, streams, snapshotMap);
+        callback(err, streams, snapshotMap);
       });
     } else {
       // If a versions map, get ops since requested versions

lib/client/connection.js

@@ -175,11 +175,7 @@ Connection.prototype.bindToSocket = function(socket) {
 Connection.prototype.handleMessage = function(message) {
   var err = null;
   if (message.error) {
-    // wrap in Error object so can be passed through event emitters
-    err = new Error(message.error.message);
-    err.code = message.error.code;
-    // Add the message data to the error object for more context
-    err.data = message;
+    err = wrapErrorData(message.error, message);
     delete message.error;
   }
   // Switch on the message action. Most messages are for documents and are
@@ -233,11 +229,11 @@ Connection.prototype.handleMessage = function(message) {
       return;
 
     case 'bf':
-      return this._handleBulkMessage(message, '_handleFetch');
+      return this._handleBulkMessage(err, message, '_handleFetch');
     case 'bs':
-      return this._handleBulkMessage(message, '_handleSubscribe');
+      return this._handleBulkMessage(err, message, '_handleSubscribe');
     case 'bu':
-      return this._handleBulkMessage(message, '_handleUnsubscribe');
+      return this._handleBulkMessage(err, message, '_handleUnsubscribe');
 
     case 'nf':
     case 'nt':
@@ -265,22 +261,43 @@ Connection.prototype.handleMessage = function(message) {
   }
 };
 
-Connection.prototype._handleBulkMessage = function(message, method) {
+function wrapErrorData(errorData, fullMessage) {
+  // wrap in Error object so can be passed through event emitters
+  var err = new Error(errorData.message);
+  err.code = errorData.code;
+  if (fullMessage) {
+    // Add the message data to the error object for more context
+    err.data = fullMessage;
+  }
+  return err;
+}
+
+Connection.prototype._handleBulkMessage = function(err, message, method) {
   if (message.data) {
     for (var id in message.data) {
+      var dataForId = message.data[id];
       var doc = this.getExisting(message.c, id);
-      if (doc) doc[method](message.error, message.data[id]);
+      if (doc) {
+        if (err) {
+          doc[method](err);
+        } else if (dataForId.error) {
+          // Bulk reply snapshot-specific errorr - see agent.js getMapResult
+          doc[method](wrapErrorData(dataForId.error));
+        } else {
+          doc[method](null, dataForId);
+        }
+      }
     }
   } else if (Array.isArray(message.b)) {
     for (var i = 0; i < message.b.length; i++) {
       var id = message.b[i];
       var doc = this.getExisting(message.c, id);
-      if (doc) doc[method](message.error);
+      if (doc) doc[method](err);
     }
   } else if (message.b) {
     for (var id in message.b) {
       var doc = this.getExisting(message.c, id);
-      if (doc) doc[method](message.error);
+      if (doc) doc[method](err);
     }
   } else {
     logger.error('Invalid bulk message', message);

lib/client/doc.js

@@ -246,6 +246,14 @@ Doc.prototype._emitNothingPending = function() {
 // **** Helpers for network messages
 
 Doc.prototype._emitResponseError = function(err, callback) {
+  if (err && err.code === ERROR_CODE.ERR_SNAPSHOT_READ_SILENT_REJECTION) {
+    this.wantSubscribe = false;
+    if (callback) {
+      callback();
+    }
+    this._emitNothingPending();
+    return;
+  }
   if (callback) {
     callback(err);
     this._emitNothingPending();

lib/error.js

@@ -41,6 +41,25 @@ ShareDBError.CODES = {
   ERR_OT_OP_NOT_PROVIDED: 'ERR_OT_OP_NOT_PROVIDED',
   ERR_PROTOCOL_VERSION_NOT_SUPPORTED: 'ERR_PROTOCOL_VERSION_NOT_SUPPORTED',
   ERR_QUERY_EMITTER_LISTENER_NOT_ASSIGNED: 'ERR_QUERY_EMITTER_LISTENER_NOT_ASSIGNED',
+  /**
+   * A special error that a "readSnapshots" middleware implementation can use to indicate that it
+   * wishes for the ShareDB client to treat it as a silent rejection, not passing the error back to
+   * user code.
+   *
+   * For subscribes, the ShareDB client will still cancel the document subscription.
+   */
+  ERR_SNAPSHOT_READ_SILENT_REJECTION: 'ERR_SNAPSHOT_READ_SILENT_REJECTION',
+  /**
+   * A "readSnapshots" middleware rejected the reads of specific snapshots.
+   *
+   * This error code is mostly for server use and generally will not be encountered on the client.
+   * Instead, each specific doc that encountered an error will receive its specific error.
+   *
+   * The one exception is for queries, where a "readSnapshots" rejection of specific snapshots will
+   * cause the client to receive this error for the whole query, since queries don't support
+   * doc-specific errors.
+   */
+  ERR_SNAPSHOT_READS_REJECTED: 'ERR_SNAPSHOT_READS_REJECTED',
   ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND: 'ERR_SUBMIT_TRANSFORM_OPS_NOT_FOUND',
   ERR_TYPE_CANNOT_BE_PROJECTED: 'ERR_TYPE_CANNOT_BE_PROJECTED',
   ERR_UNKNOWN_ERROR: 'ERR_UNKNOWN_ERROR'