implemented auth utilities for yjs

src/Connector.js

 /* @flow */
 'use strict'
 
+function canRead (auth) { return auth === 'read' || auth === 'write' }
+function canWrite (auth) { return auth === 'write' }
+
 module.exports = function (Y/* :any */) {
   class AbstractConnector {
     /* ::
@@ -54,6 +57,8 @@ module.exports = function (Y/* :any */) {
       this.syncStep2 = Promise.resolve()
       this.broadcastOpBuffer = []
       this.protocolVersion = 11
+      this.authInfo = opts.auth || null
+      this.checkAuth = opts.checkAuth || function () { return Promise.resolve('write') } // default is everyone has write access
     }
     reconnect () {
     }
@@ -87,6 +92,9 @@ module.exports = function (Y/* :any */) {
     onUserEvent (f) {
       this.userEventListeners.push(f)
     }
+    removeUserEventListener (f) {
+      this.userEventListeners = this.userEventListeners.filter(g => { f !== g })
+    }
     userLeft (user) {
       if (this.connections[user] != null) {
         delete this.connections[user]
@@ -163,7 +171,8 @@ module.exports = function (Y/* :any */) {
             type: 'sync step 1',
             stateSet: stateSet,
             deleteSet: deleteSet,
-            protocolVersion: conn.protocolVersion
+            protocolVersion: conn.protocolVersion,
+            auth: conn.authInfo
           })
         })
       } else {
@@ -217,7 +226,7 @@ module.exports = function (Y/* :any */) {
     */
     receiveMessage (sender/* :UserId */, message/* :Message */) {
       if (sender === this.userId) {
-        return
+        return Promise.resolve()
       }
       if (this.debug) {
         console.log(`receive ${sender} -> ${this.userId}: ${message.type}`, JSON.parse(JSON.stringify(message))) // eslint-disable-line
@@ -232,91 +241,118 @@ module.exports = function (Y/* :any */) {
           type: 'sync stop',
           protocolVersion: this.protocolVersion
         })
-        return
+        return Promise.reject('Incompatible protocol version')
       }
-      if (message.type === 'sync step 1') {
-        let conn = this
-        let m = message
-        this.y.db.requestTransaction(function *() {
-          var currentStateSet = yield* this.getStateSet()
-          yield* this.applyDeleteSet(m.deleteSet)
-
-          var ds = yield* this.getDeleteSet()
-          var ops = yield* this.getOperations(m.stateSet)
-          conn.send(sender, {
-            type: 'sync step 2',
-            os: ops,
-            stateSet: currentStateSet,
-            deleteSet: ds,
-            protocolVersion: this.protocolVersion
-          })
-          if (this.forwardToSyncingClients) {
-            conn.syncingClients.push(sender)
-            setTimeout(function () {
-              conn.syncingClients = conn.syncingClients.filter(function (cli) {
-                return cli !== sender
-              })
-              conn.send(sender, {
-                type: 'sync done'
-              })
-            }, 5000) // TODO: conn.syncingClientDuration)
-          } else {
-            conn.send(sender, {
-              type: 'sync done'
+      if (message.auth != null && this.connections[sender] != null) {
+        // authenticate using auth in message
+        var auth = this.checkAuth(message.auth, this.y)
+        this.connections[sender].auth = auth
+        auth.then(auth => {
+          for (var f of this.userEventListeners) {
+            f({
+              action: 'userAuthenticated',
+              user: sender,
+              auth: auth
             })
           }
-          conn._setSyncedWith(sender)
         })
-      } else if (message.type === 'sync step 2') {
-        let conn = this
-        var broadcastHB = !this.broadcastedHB
-        this.broadcastedHB = true
-        var db = this.y.db
-        var defer = {}
-        defer.promise = new Promise(function (resolve) {
-          defer.resolve = resolve
-        })
-        this.syncStep2 = defer.promise
-        let m /* :MessageSyncStep2 */ = message
-        db.requestTransaction(function * () {
-          yield* this.applyDeleteSet(m.deleteSet)
-          this.store.apply(m.os)
-          db.requestTransaction(function * () {
-            var ops = yield* this.getOperations(m.stateSet)
-            if (ops.length > 0) {
-              if (!broadcastHB) { // TODO: consider to broadcast here..
+      } else if (this.connections[sender] != null && this.connections[sender].auth == null) {
+        // authenticate without otherwise
+        this.connections[sender].auth = this.checkAuth(null, this.y)
+      }
+      if (this.connections[sender] != null && this.connections[sender].auth != null) {
+        return this.connections[sender].auth.then((auth) => {
+          if (message.type === 'sync step 1' && canRead(auth)) {
+            let conn = this
+            let m = message
+
+            this.y.db.requestTransaction(function *() {
+              var currentStateSet = yield* this.getStateSet()
+              if (canWrite(auth)) {
+                yield* this.applyDeleteSet(m.deleteSet)
+              }
+
+              var ds = yield* this.getDeleteSet()
+              var ops = yield* this.getOperations(m.stateSet)
+              conn.send(sender, {
+                type: 'sync step 2',
+                os: ops,
+                stateSet: currentStateSet,
+                deleteSet: ds,
+                protocolVersion: this.protocolVersion,
+                auth: this.authInfo
+              })
+              if (this.forwardToSyncingClients) {
+                conn.syncingClients.push(sender)
+                setTimeout(function () {
+                  conn.syncingClients = conn.syncingClients.filter(function (cli) {
+                    return cli !== sender
+                  })
+                  conn.send(sender, {
+                    type: 'sync done'
+                  })
+                }, 5000) // TODO: conn.syncingClientDuration)
+              } else {
                 conn.send(sender, {
-                  type: 'update',
-                  ops: ops
+                  type: 'sync done'
                 })
-              } else {
-                // broadcast only once!
-                conn.broadcastOps(ops)
+              }
+              conn._setSyncedWith(sender)
+            })
+          } else if (message.type === 'sync step 2' && canWrite(auth)) {
+            let conn = this
+            var broadcastHB = !this.broadcastedHB
+            this.broadcastedHB = true
+            var db = this.y.db
+            var defer = {}
+            defer.promise = new Promise(function (resolve) {
+              defer.resolve = resolve
+            })
+            this.syncStep2 = defer.promise
+            let m /* :MessageSyncStep2 */ = message
+            db.requestTransaction(function * () {
+              yield* this.applyDeleteSet(m.deleteSet)
+              this.store.apply(m.os)
+              db.requestTransaction(function * () {
+                var ops = yield* this.getOperations(m.stateSet)
+                if (ops.length > 0) {
+                  if (!broadcastHB) { // TODO: consider to broadcast here..
+                    conn.send(sender, {
+                      type: 'update',
+                      ops: ops
+                    })
+                  } else {
+                    // broadcast only once!
+                    conn.broadcastOps(ops)
+                  }
+                }
+                defer.resolve()
+              })
+            })
+          } else if (message.type === 'sync done') {
+            var self = this
+            this.syncStep2.then(function () {
+              self._setSyncedWith(sender)
+            })
+          } else if (message.type === 'update' && canWrite(auth)) {
+            if (this.forwardToSyncingClients) {
+              for (var client of this.syncingClients) {
+                this.send(client, message)
               }
             }
-            defer.resolve()
-          })
-        })
-      } else if (message.type === 'sync done') {
-        var self = this
-        this.syncStep2.then(function () {
-          self._setSyncedWith(sender)
-        })
-      } else if (message.type === 'update') {
-        if (this.forwardToSyncingClients) {
-          for (var client of this.syncingClients) {
-            this.send(client, message)
-          }
-        }
-        if (this.y.db.forwardAppliedOperations) {
-          var delops = message.ops.filter(function (o) {
-            return o.struct === 'Delete'
-          })
-          if (delops.length > 0) {
-            this.broadcastOps(delops)
+            if (this.y.db.forwardAppliedOperations) {
+              var delops = message.ops.filter(function (o) {
+                return o.struct === 'Delete'
+              })
+              if (delops.length > 0) {
+                this.broadcastOps(delops)
+              }
+            }
+            this.y.db.apply(message.ops)
           }
-        }
-        this.y.db.apply(message.ops)
+        })
+      } else {
+        return Promise.reject('Unable to deliver message')
       }
     }
     _setSyncedWith (user) {

src/Connectors/Test.js

@@ -24,11 +24,21 @@ module.exports = function (Y) {
       }
     },
     whenTransactionsFinished: function () {
-      var ps = []
-      for (var name in this.users) {
-        ps.push(this.users[name].y.db.whenTransactionsFinished())
-      }
-      return Promise.all(ps)
+      var self = this
+      return new Promise (function (resolve, reject) {
+        // The connector first has to send the messages to the db.
+        // Wait for the checkAuth-function to resolve
+        // The test lib only has a simple checkAuth function: `() => Promise.resolve()`
+        // Just add a function to the event-queue, in order to wait for the event.
+        // TODO: this may be buggy in test applications (but it isn't be for real-life apps)
+        setTimeout(function () {
+          var ps = []
+          for (var name in self.users) {
+            ps.push(self.users[name].y.db.whenTransactionsFinished())
+          }
+          Promise.all(ps).then(resolve, reject)
+        }, 0)
+      })
     },
     flushOne: function flushOne () {
       var bufs = []
@@ -54,8 +64,9 @@ module.exports = function (Y) {
           delete buff[sender]
         }
         var user = globalRoom.users[userId]
-        user.receiveMessage(m[0], m[1])
-        return user.y.db.whenTransactionsFinished()
+        return user.receiveMessage(m[0], m[1]).then(function () {
+          return user.y.db.whenTransactionsFinished() 
+        }, function () {})
       } else {
         return false
       }
@@ -72,16 +83,14 @@ module.exports = function (Y) {
             }
             globalRoom.whenTransactionsFinished().then(nextFlush)
           } else {
-            setTimeout(function () {
-              var c = globalRoom.flushOne()
-              if (c) {
-                c.then(function () {
-                  globalRoom.whenTransactionsFinished().then(nextFlush)
-                })
-              } else {
-                resolve()
-              }
-            }, 0)
+            var c = globalRoom.flushOne()
+            if (c) {
+              c.then(function () {
+                globalRoom.whenTransactionsFinished().then(nextFlush)
+              })
+            } else {
+              resolve()
+            }
           }
         }
         globalRoom.whenTransactionsFinished().then(nextFlush)
@@ -107,7 +116,7 @@ module.exports = function (Y) {
       this.syncingClientDuration = 0
     }
     receiveMessage (sender, m) {
-      super.receiveMessage(sender, JSON.parse(JSON.stringify(m)))
+      return super.receiveMessage(sender, JSON.parse(JSON.stringify(m)))
     }
     send (userId, message) {
       var buffer = globalRoom.buffers[userId]
@@ -154,7 +163,7 @@ module.exports = function (Y) {
           if (buff[sender].length === 0) {
             delete buff[sender]
           }
-          this.receiveMessage(m[0], m[1])
+          yield this.receiveMessage(m[0], m[1])
         }
         yield self.whenTransactionsFinished()
       })